flow runner update --from FILE: the runner's side of runner updates 85275e1ef3b7

//! Runner updates, the runner's side (docs/jetstream.md §8.2, #85): check

//! a new `flow` build, stage it beside the running one, swap it in keeping

//! the old one as `flow.prev`, and have the running runner leave with

//! notice so its unit starts the new build. The new build proves itself by

//! reaching the hub; three failed starts, or ten minutes without the hub,

//! put `flow.prev` back.

use anyhow::{bail, Context, Result};

use serde::{Deserialize, Serialize};

use std::path::{Path, PathBuf};

use std::time::Duration;

/// The staged build, beside the installed one.

pub const NEW: &str = "flow.new";

/// The build an update replaced, kept to go back to.

pub const PREV: &str = "flow.prev";

/// In the runner's home: the update not yet proven.

const PENDING: &str = "update.json";

/// In the runner's home: every update and rollback, a JSON line each.

const LOG: &str = "updates.log";

/// In the runner's home: the running runner's process id.

const PID: &str = "runner.pid";

/// Starts of an unproven build after which the previous one comes back.

pub const MAX_STARTS: u32 = 3;

/// How long a new build has to reach the hub.

pub const PROVE_WITHIN: Duration = Duration::from_secs(10 * 60);

/// An update swapped in and not yet proven.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Pending {

    pub from: String,

    pub to: String,

    /// The installed binary the update replaced.

    pub exe: PathBuf,

    pub at_ms: u64,

    /// How long the running runner had to finish what it held.

    pub notice_secs: u64,

    /// Starts of the new build so far.

    pub starts: u32,

}

/// What [`install`] did, or would do.

#[derive(Debug, PartialEq, Eq)]

pub struct Installed {

    pub from: String,

    pub to: String,

    pub exe: PathBuf,

    /// The running runner told to leave, by process id.

    pub told: Option<u32>,

}

/// The version a `flow --version` line names: `flow 0.3.1 (build)` → `0.3.1`.

pub fn parse_version(line: &str) -> Option<String> {

    let v = line.trim().strip_prefix("flow ")?.split_whitespace().next()?;

    v.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())).then(|| v.to_string())

}

/// Whether version `a` is older than `b`, part by part.

pub fn older(a: &str, b: &str) -> bool {

    let parts = |v: &str| v.split('.').map(|p| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>();

    parts(a) < parts(b)

}

/// The ELF machine number of this build's architecture.

fn machine_here() -> Option<u16> {

    Some(match std::env::consts::ARCH {

        "x86_64" => 62,

        "aarch64" => 183,

        "riscv64" => 243,

        _ => return None,

    })

}

/// Refuse a file that is not a 64-bit little-endian ELF executable for this

/// machine: a build for another target would not start, and the rollback

/// would only notice after three tries.

pub fn check_target(bytes: &[u8]) -> Result<()> {

    if bytes.len() < 20 || &bytes[..4] != b"\x7fELF" {

        bail!("not an ELF executable");

    }

    if bytes[4] != 2 || bytes[5] != 1 {

        bail!("not a 64-bit little-endian build");

    }

    let machine = u16::from_le_bytes([bytes[18], bytes[19]]);

    match machine_here() {

        Some(m) if m == machine => Ok(()),

        Some(_) => bail!("a build for another machine (ELF machine {machine}); this runner is {}", std::env::consts::ARCH),

        None => bail!("updates are not supported on {}", std::env::consts::ARCH),

    }

}

/// Run `file args…` with a time limit, returning its standard output.

fn run(file: &Path, args: &[&str], limit: Duration) -> Result<String> {

    let mut child = std::process::Command::new(file)

        .args(args)

        .stdin(std::process::Stdio::null())

        .stdout(std::process::Stdio::piped())

        .stderr(std::process::Stdio::piped())

        .spawn()

        .with_context(|| format!("starting {}", file.display()))?;

    let start = std::time::Instant::now();

    loop {

        if child.try_wait()?.is_some() {

            break;

        }

        if start.elapsed() > limit {

            let _ = child.kill();

            let _ = child.wait();

            bail!("`{} {}` did not finish within {} s", file.display(), args.join(" "), limit.as_secs());

        }

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

    }

    let out = child.wait_with_output()?;

    if !out.status.success() {

        bail!("`{} {}` failed ({}): {}", file.display(), args.join(" "), out.status, String::from_utf8_lossy(&out.stderr).trim());

    }

    Ok(String::from_utf8_lossy(&out.stdout).into_owned())

}

/// Check a build as the runner would run it: this machine's target, it

/// starts and names its version, and it finds the tiers this home offers.

/// Returns its version.

pub fn check(file: &Path, home: &Path) -> Result<String> {

    let mut head = vec![0u8; 64];

    let n = std::io::Read::read(&mut std::fs::File::open(file).with_context(|| format!("opening {}", file.display()))?, &mut head)?;

    check_target(&head[..n])?;

    let limit = Duration::from_secs(30);

    let line = run(file, &["--version"], limit)?;

    let version = parse_version(&line).with_context(|| format!("{} does not say it is flow: {:?}", file.display(), line.trim()))?;

    let home_s = home.to_string_lossy();

    let tiers = run(file, &["--home", &home_s, "tiers"], limit)?;

    if !tiers.lines().any(|l| l.ends_with(": available")) {

        bail!("flow {version} finds no tier it can run here:\n{}", tiers.trim());

    }

    Ok(version)

}

fn now_ms() -> u64 {

    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)

}

fn log(home: &Path, event: &str, from: &str, to: &str, why: &str) {

    use std::io::Write;

    let line = serde_json::json!({"at_ms": now_ms(), "event": event, "from": from, "to": to, "why": why});

    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(home.join(LOG)) {

        let _ = writeln!(f, "{line}");

    }

}

/// The update not yet proven, if any.

pub fn pending(home: &Path) -> Option<Pending> {

    serde_json::from_slice(&std::fs::read(home.join(PENDING)).ok()?).ok()

}

fn save(home: &Path, p: &Pending) -> Result<()> {

    let tmp = home.join(format!("{PENDING}.tmp"));

    std::fs::write(&tmp, serde_json::to_vec_pretty(p)?)?;

    std::fs::rename(&tmp, home.join(PENDING))?;

    Ok(())

}

/// Put `staged` in place of `exe`, keeping `exe` as [`PREV`] beside it.

/// `exe` is never missing: it is copied aside, then replaced in one rename.

pub fn swap(exe: &Path, staged: &Path) -> Result<()> {

    let prev = exe.with_file_name(PREV);

    let _ = std::fs::remove_file(&prev);

    if std::fs::hard_link(exe, &prev).is_err() {

        std::fs::copy(exe, &prev).with_context(|| format!("keeping {} as {}", exe.display(), prev.display()))?;

    }

    std::fs::rename(staged, exe).with_context(|| format!("putting {} in place of {}", staged.display(), exe.display()))?;

    Ok(())

}

/// Put [`PREV`] back in place of `exe`.

pub fn restore(exe: &Path) -> Result<()> {

    let prev = exe.with_file_name(PREV);

    if !prev.is_file() {

        bail!("{} is not there to go back to", prev.display());

    }

    std::fs::rename(&prev, exe).with_context(|| format!("putting {} back", prev.display()))?;

    Ok(())

}

/// Check `file`, stage it beside `exe`, and unless `dry_run` swap it in and

/// tell the running runner (if any) to leave within `notice`, finishing

/// what fits and handing back the rest, so its unit starts the new build.

pub fn install(home: &crate::home::Home, exe: &Path, file: &Path, notice: Duration, dry_run: bool, allow_older: bool) -> Result<Installed> {

    if home.config.runner.as_ref().and_then(|r| r.update.as_ref()).is_some_and(|u| !u.allow) {

        bail!("flow.toml says [runner.update] allow = false: this runner takes no updates");

    }

    if let Some(p) = pending(&home.root) {

        bail!("the update from {} to {} is not proven yet; wait for it (or for its rollback)", p.from, p.to);

    }

    let from = env!("CARGO_PKG_VERSION").to_string();

    let staged = exe.with_file_name(NEW);

    let _ = std::fs::remove_file(&staged);

    std::fs::copy(file, &staged).with_context(|| format!("staging {} as {}", file.display(), staged.display()))?;

    {

        use std::os::unix::fs::PermissionsExt;

        std::fs::set_permissions(&staged, std::fs::Permissions::from_mode(0o755))?;

    }

    let checked = check(&staged, &home.root).and_then(|to| {

        if older(&to, &from) && !allow_older {

            bail!("flow {to} is older than this runner's {from} (--allow-older to go back)");

        }

        Ok(to)

    });

    let to = match checked {

        Ok(v) => v,

        Err(e) => {

            let _ = std::fs::remove_file(&staged);

            return Err(e);

        }

    };

    if dry_run {

        let _ = std::fs::remove_file(&staged);

        return Ok(Installed { from, to, exe: exe.to_path_buf(), told: None });

    }

    swap(exe, &staged)?;

    let p =

        Pending { from: from.clone(), to: to.clone(), exe: exe.to_path_buf(), at_ms: now_ms(), notice_secs: notice.as_secs(), starts: 0 };

    save(&home.root, &p)?;

    log(&home.root, "installed", &from, &to, "flow runner update --from");

    let told = running(&home.root).filter(|pid| unsafe { libc::kill(*pid as libc::pid_t, libc::SIGUSR1) } == 0);

    Ok(Installed { from, to, exe: exe.to_path_buf(), told })

}

/// The running runner's process id, from its pid file, if that process is

/// still a `flow runner`: a stale file's pid may be another program's now,

/// and SIGUSR1 would end it. Builds before #85 write no pid file (and do

/// not listen for SIGUSR1), so they are never told.

fn running(home: &Path) -> Option<u32> {

    let pid: u32 = std::fs::read_to_string(home.join(PID)).ok()?.trim().parse().ok()?;

    let cmdline = std::fs::read(format!("/proc/{pid}/cmdline")).ok()?;

    is_runner(&cmdline).then_some(pid)

}

/// Whether a NUL-separated command line is `flow … runner` with no subcommand.

fn is_runner(cmdline: &[u8]) -> bool {

    let args: Vec<&[u8]> = cmdline.split(|b| *b == 0).filter(|a| !a.is_empty()).collect();

    let Some((argv0, rest)) = args.split_first() else { return false };

    let name = argv0.rsplit(|b| *b == b'/').next().unwrap_or(argv0);

    name.starts_with(b"flow") && rest.last() == Some(&&b"runner"[..])

}

/// Note this process as the home's runner, for `flow runner update` to tell.

pub fn note_runner(home: &Path) {

    let _ = std::fs::write(home.join(PID), std::process::id().to_string());

}

/// The runner stopped: forget its pid.

pub fn forget_runner(home: &Path) {

    if std::fs::read_to_string(home.join(PID)).is_ok_and(|p| p.trim() == std::process::id().to_string()) {

        let _ = std::fs::remove_file(home.join(PID));

    }

}

/// What a runner starting up should do about an update.

#[derive(Debug, PartialEq, Eq)]

pub enum Start {

    /// No update waits to be proven.

    Settled,

    /// The new build is on trial: prove it within [`PROVE_WITHIN`].

    Trial(Pending),

    /// It failed too often; the previous build is back and should be run.

    RolledBack(Pending),

}

/// Count this start of an unproven build; after [`MAX_STARTS`], put the

/// previous build back.

pub fn on_start(home: &Path) -> Result<Start> {

    let Some(mut p) = pending(home) else { return Ok(Start::Settled) };

    p.starts += 1;

    if p.starts > MAX_STARTS {

        roll_back(home, &p, &format!("{MAX_STARTS} starts without reaching the hub"))?;

        return Ok(Start::RolledBack(p));

    }

    save(home, &p)?;

    Ok(Start::Trial(p))

}

/// The new build reached the hub: it stays.

pub fn proven(home: &Path) {

    if let Some(p) = pending(home) {

        let _ = std::fs::remove_file(home.join(PENDING));

        log(home, "proven", &p.from, &p.to, "reached the hub");

        eprintln!("flow runner: flow {} proven (it reached the hub); flow {} stays as {PREV}", p.to, p.from);

    }

}

/// Put the previous build back and forget the update.

pub fn roll_back(home: &Path, p: &Pending, why: &str) -> Result<()> {

    restore(&p.exe)?;

    let _ = std::fs::remove_file(home.join(PENDING));

    log(home, "rolled back", &p.to, &p.from, why);

    eprintln!("flow runner: rolled back from flow {} to {} ({why})", p.to, p.from);

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn versions_are_read_and_compared() {

        assert_eq!(parse_version("flow 0.3.1 (local build)\n").as_deref(), Some("0.3.1"));

        assert_eq!(parse_version("flow 0.10.0"), Some("0.10.0".into()));

        assert_eq!(parse_version("oxy 0.6.0"), None);

        assert_eq!(parse_version("flow banana"), None);

        assert!(older("0.3.1", "0.10.0"), "by number, not by text");

        assert!(!older("0.3.1", "0.3.1"));

        assert!(older("0.3", "0.3.1"));

    }

    #[test]

    fn only_a_build_for_this_machine_passes() {

        let me = std::fs::read(std::env::current_exe().unwrap()).unwrap();

        check_target(&me[..64]).unwrap();

        assert!(check_target(b"#!/bin/sh\necho flow 9.9.9\n").unwrap_err().to_string().contains("not an ELF"));

        let mut other = me[..64].to_vec();

        other[18..20].copy_from_slice(&(if std::env::consts::ARCH == "riscv64" { 62u16 } else { 243u16 }).to_le_bytes());

        assert!(check_target(&other).unwrap_err().to_string().contains("another machine"));

    }

    #[test]

    fn a_swap_keeps_the_old_build_and_three_failed_starts_bring_it_back() {

        let dir = tempfile::tempdir().unwrap();

        let (bin, home) = (dir.path().join("bin"), dir.path().join("home"));

        std::fs::create_dir_all(&bin).unwrap();

        std::fs::create_dir_all(&home).unwrap();

        let exe = bin.join("flow");

        std::fs::write(&exe, "old").unwrap();

        std::fs::write(bin.join(NEW), "new").unwrap();

        swap(&exe, &bin.join(NEW)).unwrap();

        assert_eq!(std::fs::read_to_string(&exe).unwrap(), "new");

        assert_eq!(std::fs::read_to_string(bin.join(PREV)).unwrap(), "old");

        assert!(!bin.join(NEW).exists());

        assert_eq!(on_start(&home).unwrap(), Start::Settled, "no update, nothing to do");

        let p = Pending { from: "0.3.1".into(), to: "0.3.2".into(), exe: exe.clone(), at_ms: 1, notice_secs: 600, starts: 0 };

        save(&home, &p).unwrap();

        for n in 1..=MAX_STARTS {

            assert_eq!(on_start(&home).unwrap(), Start::Trial(Pending { starts: n, ..p.clone() }));

        }

        assert!(matches!(on_start(&home).unwrap(), Start::RolledBack(_)));

        assert_eq!(std::fs::read_to_string(&exe).unwrap(), "old", "the previous build is back");

        assert!(pending(&home).is_none());

        let log = std::fs::read_to_string(home.join(LOG)).unwrap();

        assert!(log.contains("rolled back") && log.contains("0.3.2"), "{log}");

    }

    #[test]

    fn only_a_running_runner_is_told() {

        assert!(is_runner(b"/usr/local/bin/flow\0runner\0"));

        assert!(is_runner(b"flow\0--home\0/srv/flow\0runner\0"));

        assert!(!is_runner(b"flow\0runner\0update\0--from\0x\0"), "not the update command itself");

        assert!(!is_runner(b"/usr/bin/vim\0runner\0"));

        assert!(!is_runner(b""));

        let dir = tempfile::tempdir().unwrap();

        std::fs::write(dir.path().join(PID), std::process::id().to_string()).unwrap();

        assert_eq!(running(dir.path()), None, "this test is not a flow runner");

    }

    #[test]

    fn a_proven_build_stays() {

        let dir = tempfile::tempdir().unwrap();

        let exe = dir.path().join("flow");

        std::fs::write(&exe, "new").unwrap();

        let p = Pending { from: "0.3.1".into(), to: "0.3.2".into(), exe: exe.clone(), at_ms: 1, notice_secs: 0, starts: 1 };

        save(dir.path(), &p).unwrap();

        proven(dir.path());

        assert!(pending(dir.path()).is_none());

        assert_eq!(on_start(dir.path()).unwrap(), Start::Settled);

        assert!(std::fs::read_to_string(dir.path().join(LOG)).unwrap().contains("proven"));

    }

}

- **`flow runner update --from FILE` (#85):** the runner's side of self-updating runners (docs/jetstream.md §8.2). It checks a new build (this machine's target, `--version`, a tier it can run here), keeps the installed one as `flow.prev`, swaps the new one in with a single rename, and tells the running runner (SIGUSR1, its pid in `runner.pid`) to leave within `--notice` seconds (600 by default): jobs that fit finish, the rest are handed back with their checkpoints, and the unit starts the new build. The new build proves itself by getting an answer from the hub; four starts or ten minutes without one put `flow.prev` back. `--dry-run` changes nothing; `[runner.update] allow = false` in `flow.toml` refuses every update; `updates.log` in the home records each install, proof and rollback. Spec §11.5.3.

    /// `[runner.update]`: whether this runner takes updates (#85).

    #[serde(default)]

    pub update: Option<UpdateConfig>,

}

/// `[runner.update]` in flow.toml (docs/jetstream.md §8.2): the runner's

/// last word on updates, whatever the hub's policy says.

#[derive(Clone, Debug, Deserialize)]

pub struct UpdateConfig {

    /// `false` refuses every update, so a site can hold a machine back.

    #[serde(default = "yes")]

    pub allow: bool,

}

fn yes() -> bool {

    true

pub mod update;

    /// Install a new flow build in place of this one (docs/jetstream.md

    /// §8.2): check it, keep this one as flow.prev, and have the running

    /// runner leave with notice so its unit starts the new build. A build

    /// that does not reach the hub goes back to flow.prev by itself

    Update {

        /// The new flow binary

        #[arg(long)]

        from: PathBuf,

        /// Seconds the running runner has to finish what it holds; jobs

        /// that will not finish in time are handed back with their checkpoints

        #[arg(long, default_value_t = 600)]

        notice: u64,

        /// Check the build and say what would happen, changing nothing

        #[arg(long)]

        dry_run: bool,

        /// Install a build older than this one

        #[arg(long)]

        allow_older: bool,

        /// The installed binary to replace (default: this one)

        #[arg(long)]

        exe: Option<PathBuf>,

    },

            Ok(())

        }

        Cmd::Runner { cmd: Some(RunnerCmd::Update { from, notice, dry_run, allow_older, exe }) } => {

            let home = Home::locate(cli.home.as_deref())?;

            let exe = match exe {

                Some(e) => e,

                None => std::env::current_exe().context("finding this flow binary")?,

            };

            let done = flow::update::install(&home, &exe, &from, Duration::from_secs(notice), dry_run, allow_older)?;

            if dry_run {

                println!(

                    "flow {} checks out: it would replace flow {} at {} (dry run: nothing changed)",

                    done.to,

                    done.from,

                    done.exe.display()

                );

                return Ok(());

            }

            println!("flow {} installed at {}; flow {} kept as {}", done.to, done.exe.display(), done.from, flow::update::PREV);

            match done.told {

                Some(pid) => println!("the runner (pid {pid}) leaves within {notice} s, then its unit starts flow {}", done.to),

                None => println!(

                    "no runner of this home was told (none runs, or one from before runner updates, which does not listen): \

                     restart its unit, e.g. `systemctl restart --no-block flow-runner`; it drains and starts flow {}",

                    done.to

                ),

            }

            // A new build on trial that keeps failing to start goes back to

            // the one before, which then runs in its place (#85).

            if let flow::update::Start::RolledBack(p) = flow::update::on_start(&home.root)? {

                use std::os::unix::process::CommandExt;

                let err = std::process::Command::new(&p.exe).args(std::env::args_os().skip(1)).exec();

                bail!("starting flow {} again after the rollback: {err}", p.from);

            }

    crate::update::note_runner(&home.root);

    // `flow runner update` swapped a new build in and asks this one to

    // leave with the notice it gave (#85).

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

    unsafe { libc::signal(libc::SIGUSR1, handler) };

    if let Some(p) = crate::update::pending(&home.root) {

        // On trial: a build that does not reach the hub in time goes back.

        let root = home.root.clone();

        std::thread::spawn(move || {

            std::thread::sleep(crate::update::PROVE_WITHIN);

            if crate::update::pending(&root).is_some_and(|now| now.to == p.to && now.at_ms == p.at_ms) {

                let why = format!("did not reach the hub within {} minutes", crate::update::PROVE_WITHIN.as_secs() / 60);

                if crate::update::roll_back(&root, &p, &why).is_ok() {

                    std::process::exit(1);

                }

            }

        });

    }

    let served = serve_until_drained(home);

    crate::update::forget_runner(&home.root);

    served

/// Set by SIGUSR1: a new build is in place; leave for it.

static UPDATE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

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

    UPDATE.store(true, std::sync::atomic::Ordering::SeqCst);

}

        let mut on_trial = crate::update::pending(&home.root).is_some();

            if UPDATE.swap(false, std::sync::atomic::Ordering::SeqCst) {

                let p = crate::update::pending(&home.root);

                let notice = Duration::from_secs(p.as_ref().map_or(0, |p| p.notice_secs));

                let to = p.map_or_else(|| "a new build".to_string(), |p| format!("flow {}", p.to));

                eprintln!("flow runner {name}: updating to {to}: leaving within {} s", notice.as_secs());

                leave_within(&format!("updating to {to}"), notice);

            }

            let answer = hub.claim(&req);

            if on_trial && answer.is_ok() {

                crate::update::proven(&home.root);

                on_trial = false;

            }

            match answer {

update --from FILE`): check, stage, drain, swap, prove, roll back (done,

#85: spec §11.5.3); (2) the

### 11.5.3 Updating a runner (oxygenflow#85)

`flow runner update --from FILE` installs a new `flow` build in place of

the one it runs as (`--exe` names another), the runner's side of

self-updating runners (docs/jetstream.md §8.2):

1. **Refused** when the home's `flow.toml` says `[runner.update] allow =

   false`, when an earlier update is not yet proven, or when the build is

   older than the installed one (unless `--allow-older`).

2. **Checked:** a 64-bit little-endian ELF for this machine; it names

   itself (`--version`, `flow X.Y.Z …`) and finds a tier it can run here

   (`tiers` against the runner's home), each within 30 seconds.

3. **Staged and swapped:** copied beside the installed binary as

   `flow.new` and checked there; the installed one is kept as

   `flow.prev` (a hard link, else a copy), then `flow.new` replaces it in

   one rename, so `flow` is never missing. `--dry-run` stops before the

   swap and changes nothing.

4. **The running runner leaves:** a runner writes its pid to

   `runner.pid` in its home, and the update sends it SIGUSR1 (after

   checking that pid is still a `flow … runner`). It then leaves with

   `--notice` (default 600 s, §11.5.2): jobs expected to finish in time

   finish, the rest are handed back with their checkpoints, and it exits

   0, so its unit's `Restart=always` starts the new build. A runner from

   before #85 is not told; restarting its unit drains it as before.

5. **Proven or rolled back:** the update waits in `update.json` until the

   new build's first claim gets an answer from the hub. Each start counts;

   on the fourth start without that answer, or after ten minutes running

   without it, `flow.prev` is put back (and on a start, run in its place).

   `updates.log` in the home records every install, proof and rollback.