update.rs 59.5%

166 of 279 lines run, 30 of 97 functions · run #188 on main · run missed

//! 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 !cfg!(unix) {
        bail!("runners update themselves on Linux only, so far");
    }
    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()))?;
    crate::portable::set_mode(&staged, 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| tell(*pid));
    Ok(Installed { from, to, exe: exe.to_path_buf(), told })
}

/// Ask runner `pid` to leave for the new build (SIGUSR1).
#[cfg(unix)]
fn tell(pid: u32) -> bool {
    unsafe { libc::kill(pid as libc::pid_t, libc::SIGUSR1) == 0 }
}

#[cfg(not(unix))]
fn tell(_pid: u32) -> bool {
    false
}

/// 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))
}

/// In the runner's home: what to tell the hub once it answers.
const NOTE: &str = "update.note";

/// Keep `note` to tell the hub (a proof, a rollback) once it answers.
fn note(home: &Path, note: &str) {
    let _ = std::fs::write(home.join(NOTE), note);
}

/// What there is to tell the hub, once.
pub fn take_note(home: &Path) -> Option<String> {
    let n = std::fs::read_to_string(home.join(NOTE)).ok()?;
    let _ = std::fs::remove_file(home.join(NOTE));
    Some(n).filter(|n| !n.trim().is_empty())
}

/// 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");
        note(home, &format!("runs flow {} (was {}); proven", p.to, p.from));
        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);
    note(home, &format!("rolled back from flow {} to {}: {why}", p.to, p.from));
    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"));
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! Runner updates, the runner's side (docs/jetstream.md §8.2, #85): check
2		//! a new `flow` build, stage it beside the running one, swap it in keeping
3		//! the old one as `flow.prev`, and have the running runner leave with
4		//! notice so its unit starts the new build. The new build proves itself by
5		//! reaching the hub; three failed starts, or ten minutes without the hub,
6		//! put `flow.prev` back.
7
8		use anyhow::{bail, Context, Result};
9		use serde::{Deserialize, Serialize};
10		use std::path::{Path, PathBuf};
11		use std::time::Duration;
12
13		/// The staged build, beside the installed one.
14		pub const NEW: &str = "flow.new";
15		/// The build an update replaced, kept to go back to.
16		pub const PREV: &str = "flow.prev";
17		/// In the runner's home: the update not yet proven.
18		const PENDING: &str = "update.json";
19		/// In the runner's home: every update and rollback, a JSON line each.
20		const LOG: &str = "updates.log";
21		/// In the runner's home: the running runner's process id.
22		const PID: &str = "runner.pid";
23		/// Starts of an unproven build after which the previous one comes back.
24		pub const MAX_STARTS: u32 = 3;
25		/// How long a new build has to reach the hub.
26		pub const PROVE_WITHIN: Duration = Duration::from_secs(10 * 60);
27
28		/// An update swapped in and not yet proven.
29		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
30		pub struct Pending {
31		pub from: String,
32		pub to: String,
33		/// The installed binary the update replaced.
34		pub exe: PathBuf,
35		pub at_ms: u64,
36		/// How long the running runner had to finish what it held.
37		pub notice_secs: u64,
38		/// Starts of the new build so far.
39		pub starts: u32,
40		}
41
42		/// What [`install`] did, or would do.
43		#[derive(Debug, PartialEq, Eq)]
44		pub struct Installed {
45		pub from: String,
46		pub to: String,
47		pub exe: PathBuf,
48		/// The running runner told to leave, by process id.
49		pub told: Option<u32>,
50		}
51
52		/// The version a `flow --version` line names: `flow 0.3.1 (build)` → `0.3.1`.
53	4×	pub fn parse_version(line: &str) -> Option<String> {
54	4×	let v = line.trim().strip_prefix("flow ")?.split_whitespace().next()?;
55	8×	v.split('.').all(\|p\| !p.is_empty() && p.bytes().all(\|b\| b.is_ascii_digit())).then(\|\| v.to_string())
56	4×	}
57
58		/// Whether version `a` is older than `b`, part by part.
59	3×	pub fn older(a: &str, b: &str) -> bool {
60	17×	let parts = \|v: &str\| v.split('.').map(\|p\| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>();
61	3×	parts(a) < parts(b)
62	3×	}
63
64		/// The ELF machine number of this build's architecture.
65	2×	fn machine_here() -> Option<u16> {
66	2×	Some(match std::env::consts::ARCH {
67	2×	"x86_64" => 62,
68	0×	"aarch64" => 183,
69	0×	"riscv64" => 243,
70	0×	_ => return None,
71		})
72	2×	}
73
74		/// Refuse a file that is not a 64-bit little-endian ELF executable for this
75		/// machine: a build for another target would not start, and the rollback
76		/// would only notice after three tries.
77	3×	pub fn check_target(bytes: &[u8]) -> Result<()> {
78	3×	if bytes.len() < 20 \|\| &bytes[..4] != b"\x7fELF" {
79	1×	bail!("not an ELF executable");
80	2×	}
81	2×	if bytes[4] != 2 \|\| bytes[5] != 1 {
82	0×	bail!("not a 64-bit little-endian build");
83	2×	}
84	2×	let machine = u16::from_le_bytes([bytes[18], bytes[19]]);
85	2×	match machine_here() {
86	2×	Some(m) if m == machine => Ok(()),
87	1×	Some(_) => bail!("a build for another machine (ELF machine {machine}); this runner is {}", std::env::consts::ARCH),
88	0×	None => bail!("updates are not supported on {}", std::env::consts::ARCH),
89		}
90	3×	}
91
92		/// Run `file args…` with a time limit, returning its standard output.
93	0×	fn run(file: &Path, args: &[&str], limit: Duration) -> Result<String> {
94	0×	let mut child = std::process::Command::new(file)
95	0×	.args(args)
96	0×	.stdin(std::process::Stdio::null())
97	0×	.stdout(std::process::Stdio::piped())
98	0×	.stderr(std::process::Stdio::piped())
99	0×	.spawn()
100	0×	.with_context(\|\| format!("starting {}", file.display()))?;
101	0×	let start = std::time::Instant::now();
102		loop {
103	0×	if child.try_wait()?.is_some() {
104	0×	break;
105	0×	}
106	0×	if start.elapsed() > limit {
107	0×	let _ = child.kill();
108	0×	let _ = child.wait();
109	0×	bail!("`{} {}` did not finish within {} s", file.display(), args.join(" "), limit.as_secs());
110	0×	}
111	0×	std::thread::sleep(Duration::from_millis(50));
112		}
113	0×	let out = child.wait_with_output()?;
114	0×	if !out.status.success() {
115	0×	bail!("`{} {}` failed ({}): {}", file.display(), args.join(" "), out.status, String::from_utf8_lossy(&out.stderr).trim());
116	0×	}
117	0×	Ok(String::from_utf8_lossy(&out.stdout).into_owned())
118	0×	}
119
120		/// Check a build as the runner would run it: this machine's target, it
121		/// starts and names its version, and it finds the tiers this home offers.
122		/// Returns its version.
123	0×	pub fn check(file: &Path, home: &Path) -> Result<String> {
124	0×	let mut head = vec![0u8; 64];
125	0×	let n = std::io::Read::read(&mut std::fs::File::open(file).with_context(\|\| format!("opening {}", file.display()))?, &mut head)?;
126	0×	check_target(&head[..n])?;
127	0×	let limit = Duration::from_secs(30);
128	0×	let line = run(file, &["--version"], limit)?;
129	0×	let version = parse_version(&line).with_context(\|\| format!("{} does not say it is flow: {:?}", file.display(), line.trim()))?;
130	0×	let home_s = home.to_string_lossy();
131	0×	let tiers = run(file, &["--home", &home_s, "tiers"], limit)?;
132	0×	if !tiers.lines().any(\|l\| l.ends_with(": available")) {
133	0×	bail!("flow {version} finds no tier it can run here:\n{}", tiers.trim());
134	0×	}
135	0×	Ok(version)
136	0×	}
137
138	2×	fn now_ms() -> u64 {
139	2×	std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0)
140	2×	}
141
142	2×	fn log(home: &Path, event: &str, from: &str, to: &str, why: &str) {
143		use std::io::Write;
144	2×	let line = serde_json::json!({"at_ms": now_ms(), "event": event, "from": from, "to": to, "why": why});
145	2×	if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(home.join(LOG)) {
146	2×	let _ = writeln!(f, "{line}");
147	2×	}
148	2×	}
149
150		/// The update not yet proven, if any.
151	9×	pub fn pending(home: &Path) -> Option<Pending> {
152	9×	serde_json::from_slice(&std::fs::read(home.join(PENDING)).ok()?).ok()
153	9×	}
154
155	5×	fn save(home: &Path, p: &Pending) -> Result<()> {
156	5×	let tmp = home.join(format!("{PENDING}.tmp"));
157	5×	std::fs::write(&tmp, serde_json::to_vec_pretty(p)?)?;
158	5×	std::fs::rename(&tmp, home.join(PENDING))?;
159	5×	Ok(())
160	5×	}
161
162		/// Put `staged` in place of `exe`, keeping `exe` as [`PREV`] beside it.
163		/// `exe` is never missing: it is copied aside, then replaced in one rename.
164	1×	pub fn swap(exe: &Path, staged: &Path) -> Result<()> {
165	1×	let prev = exe.with_file_name(PREV);
166	1×	let _ = std::fs::remove_file(&prev);
167	1×	if std::fs::hard_link(exe, &prev).is_err() {
168	0×	std::fs::copy(exe, &prev).with_context(\|\| format!("keeping {} as {}", exe.display(), prev.display()))?;
169	1×	}
170	1×	std::fs::rename(staged, exe).with_context(\|\| format!("putting {} in place of {}", staged.display(), exe.display()))?;
171	1×	Ok(())
172	1×	}
173
174		/// Put [`PREV`] back in place of `exe`.
175	1×	pub fn restore(exe: &Path) -> Result<()> {
176	1×	let prev = exe.with_file_name(PREV);
177	1×	if !prev.is_file() {
178	0×	bail!("{} is not there to go back to", prev.display());
179	1×	}
180	1×	std::fs::rename(&prev, exe).with_context(\|\| format!("putting {} back", prev.display()))?;
181	1×	Ok(())
182	1×	}
183
184		/// Check `file`, stage it beside `exe`, and unless `dry_run` swap it in and
185		/// tell the running runner (if any) to leave within `notice`, finishing
186		/// what fits and handing back the rest, so its unit starts the new build.
187	0×	pub fn install(home: &crate::home::Home, exe: &Path, file: &Path, notice: Duration, dry_run: bool, allow_older: bool) -> Result<Installed> {
188	0×	if !cfg!(unix) {
189	0×	bail!("runners update themselves on Linux only, so far");
190	0×	}
191	0×	if home.config.runner.as_ref().and_then(\|r\| r.update.as_ref()).is_some_and(\|u\| !u.allow) {
192	0×	bail!("flow.toml says [runner.update] allow = false: this runner takes no updates");
193	0×	}
194	0×	if let Some(p) = pending(&home.root) {
195	0×	bail!("the update from {} to {} is not proven yet; wait for it (or for its rollback)", p.from, p.to);
196	0×	}
197	0×	let from = env!("CARGO_PKG_VERSION").to_string();
198	0×	let staged = exe.with_file_name(NEW);
199	0×	let _ = std::fs::remove_file(&staged);
200	0×	std::fs::copy(file, &staged).with_context(\|\| format!("staging {} as {}", file.display(), staged.display()))?;
201	0×	crate::portable::set_mode(&staged, 0o755)?;
202	0×	let checked = check(&staged, &home.root).and_then(\|to\| {
203	0×	if older(&to, &from) && !allow_older {
204	0×	bail!("flow {to} is older than this runner's {from} (--allow-older to go back)");
205	0×	}
206	0×	Ok(to)
207	0×	});
208	0×	let to = match checked {
209	0×	Ok(v) => v,
210	0×	Err(e) => {
211	0×	let _ = std::fs::remove_file(&staged);
212	0×	return Err(e);
213		}
214		};
215	0×	if dry_run {
216	0×	let _ = std::fs::remove_file(&staged);
217	0×	return Ok(Installed { from, to, exe: exe.to_path_buf(), told: None });
218	0×	}
219	0×	swap(exe, &staged)?;
220	0×	let p =
221	0×	Pending { from: from.clone(), to: to.clone(), exe: exe.to_path_buf(), at_ms: now_ms(), notice_secs: notice.as_secs(), starts: 0 };
222	0×	save(&home.root, &p)?;
223	0×	log(&home.root, "installed", &from, &to, "flow runner update --from");
224	0×	let told = running(&home.root).filter(\|pid\| tell(*pid));
225	0×	Ok(Installed { from, to, exe: exe.to_path_buf(), told })
226	0×	}
227
228		/// Ask runner `pid` to leave for the new build (SIGUSR1).
229		#[cfg(unix)]
230	0×	fn tell(pid: u32) -> bool {
231	0×	unsafe { libc::kill(pid as libc::pid_t, libc::SIGUSR1) == 0 }
232	0×	}
233
234		#[cfg(not(unix))]
235		fn tell(_pid: u32) -> bool {
236		false
237		}
238
239		/// The running runner's process id, from its pid file, if that process is
240		/// still a `flow runner`: a stale file's pid may be another program's now,
241		/// and SIGUSR1 would end it. Builds before #85 write no pid file (and do
242		/// not listen for SIGUSR1), so they are never told.
243	1×	fn running(home: &Path) -> Option<u32> {
244	1×	let pid: u32 = std::fs::read_to_string(home.join(PID)).ok()?.trim().parse().ok()?;
245	1×	let cmdline = std::fs::read(format!("/proc/{pid}/cmdline")).ok()?;
246	1×	is_runner(&cmdline).then_some(pid)
247	1×	}
248
249		/// Whether a NUL-separated command line is `flow … runner` with no subcommand.
250	6×	fn is_runner(cmdline: &[u8]) -> bool {
251	230×	let args: Vec<&[u8]> = cmdline.split(\|b\| *b == 0).filter(\|a\| !a.is_empty()).collect();
252	6×	let Some((argv0, rest)) = args.split_first() else { return false };
253	39×	let name = argv0.rsplit(\|b\| *b == b'/').next().unwrap_or(argv0);
254	5×	name.starts_with(b"flow") && rest.last() == Some(&&b"runner"[..])
255	6×	}
256
257		/// Note this process as the home's runner, for `flow runner update` to tell.
258	0×	pub fn note_runner(home: &Path) {
259	0×	let _ = std::fs::write(home.join(PID), std::process::id().to_string());
260	0×	}
261
262		/// The runner stopped: forget its pid.
263	0×	pub fn forget_runner(home: &Path) {
264	0×	if std::fs::read_to_string(home.join(PID)).is_ok_and(\|p\| p.trim() == std::process::id().to_string()) {
265	0×	let _ = std::fs::remove_file(home.join(PID));
266	0×	}
267	0×	}
268
269		/// What a runner starting up should do about an update.
270		#[derive(Debug, PartialEq, Eq)]
271		pub enum Start {
272		/// No update waits to be proven.
273		Settled,
274		/// The new build is on trial: prove it within [`PROVE_WITHIN`].
275		Trial(Pending),
276		/// It failed too often; the previous build is back and should be run.
277		RolledBack(Pending),
278		}
279
280		/// Count this start of an unproven build; after [`MAX_STARTS`], put the
281		/// previous build back.
282	6×	pub fn on_start(home: &Path) -> Result<Start> {
283	6×	let Some(mut p) = pending(home) else { return Ok(Start::Settled) };
284	4×	p.starts += 1;
285	4×	if p.starts > MAX_STARTS {
286	1×	roll_back(home, &p, &format!("{MAX_STARTS} starts without reaching the hub"))?;
287	1×	return Ok(Start::RolledBack(p));
288	3×	}
289	3×	save(home, &p)?;
290	3×	Ok(Start::Trial(p))
291	6×	}
292
293		/// In the runner's home: what to tell the hub once it answers.
294		const NOTE: &str = "update.note";
295
296		/// Keep `note` to tell the hub (a proof, a rollback) once it answers.
297	2×	fn note(home: &Path, note: &str) {
298	2×	let _ = std::fs::write(home.join(NOTE), note);
299	2×	}
300
301		/// What there is to tell the hub, once.
302	0×	pub fn take_note(home: &Path) -> Option<String> {
303	0×	let n = std::fs::read_to_string(home.join(NOTE)).ok()?;
304	0×	let _ = std::fs::remove_file(home.join(NOTE));
305	0×	Some(n).filter(\|n\| !n.trim().is_empty())
306	0×	}
307
308		/// The new build reached the hub: it stays.
309	1×	pub fn proven(home: &Path) {
310	1×	if let Some(p) = pending(home) {
311	1×	let _ = std::fs::remove_file(home.join(PENDING));
312	1×	log(home, "proven", &p.from, &p.to, "reached the hub");
313	1×	note(home, &format!("runs flow {} (was {}); proven", p.to, p.from));
314	1×	eprintln!("flow runner: flow {} proven (it reached the hub); flow {} stays as {PREV}", p.to, p.from);
315	1×	}
316	1×	}
317
318		/// Put the previous build back and forget the update.
319	1×	pub fn roll_back(home: &Path, p: &Pending, why: &str) -> Result<()> {
320	1×	restore(&p.exe)?;
321	1×	let _ = std::fs::remove_file(home.join(PENDING));
322	1×	log(home, "rolled back", &p.to, &p.from, why);
323	1×	note(home, &format!("rolled back from flow {} to {}: {why}", p.to, p.from));
324	1×	eprintln!("flow runner: rolled back from flow {} to {} ({why})", p.to, p.from);
325	1×	Ok(())
326	1×	}
327
328		#[cfg(test)]
329		mod tests {
330		use super::*;
331
332		#[test]
333	1×	fn versions_are_read_and_compared() {
334	1×	assert_eq!(parse_version("flow 0.3.1 (local build)\n").as_deref(), Some("0.3.1"));
335	1×	assert_eq!(parse_version("flow 0.10.0"), Some("0.10.0".into()));
336	1×	assert_eq!(parse_version("oxy 0.6.0"), None);
337	1×	assert_eq!(parse_version("flow banana"), None);
338	1×	assert!(older("0.3.1", "0.10.0"), "by number, not by text");
339	1×	assert!(!older("0.3.1", "0.3.1"));
340	1×	assert!(older("0.3", "0.3.1"));
341	1×	}
342
343		#[test]
344	1×	fn only_a_build_for_this_machine_passes() {
345	1×	let me = std::fs::read(std::env::current_exe().unwrap()).unwrap();
346	1×	check_target(&me[..64]).unwrap();
347	1×	assert!(check_target(b"#!/bin/sh\necho flow 9.9.9\n").unwrap_err().to_string().contains("not an ELF"));
348	1×	let mut other = me[..64].to_vec();
349	1×	other[18..20].copy_from_slice(&(if std::env::consts::ARCH == "riscv64" { 62u16 } else { 243u16 }).to_le_bytes());
350	1×	assert!(check_target(&other).unwrap_err().to_string().contains("another machine"));
351	1×	}
352
353		#[test]
354	1×	fn a_swap_keeps_the_old_build_and_three_failed_starts_bring_it_back() {
355	1×	let dir = tempfile::tempdir().unwrap();
356	1×	let (bin, home) = (dir.path().join("bin"), dir.path().join("home"));
357	1×	std::fs::create_dir_all(&bin).unwrap();
358	1×	std::fs::create_dir_all(&home).unwrap();
359	1×	let exe = bin.join("flow");
360	1×	std::fs::write(&exe, "old").unwrap();
361	1×	std::fs::write(bin.join(NEW), "new").unwrap();
362	1×	swap(&exe, &bin.join(NEW)).unwrap();
363	1×	assert_eq!(std::fs::read_to_string(&exe).unwrap(), "new");
364	1×	assert_eq!(std::fs::read_to_string(bin.join(PREV)).unwrap(), "old");
365	1×	assert!(!bin.join(NEW).exists());
366
367	1×	assert_eq!(on_start(&home).unwrap(), Start::Settled, "no update, nothing to do");
368	1×	let p = Pending { from: "0.3.1".into(), to: "0.3.2".into(), exe: exe.clone(), at_ms: 1, notice_secs: 600, starts: 0 };
369	1×	save(&home, &p).unwrap();
370	3×	for n in 1..=MAX_STARTS {
371	3×	assert_eq!(on_start(&home).unwrap(), Start::Trial(Pending { starts: n, ..p.clone() }));
372		}
373	1×	assert!(matches!(on_start(&home).unwrap(), Start::RolledBack(_)));
374	1×	assert_eq!(std::fs::read_to_string(&exe).unwrap(), "old", "the previous build is back");
375	1×	assert!(pending(&home).is_none());
376	1×	let log = std::fs::read_to_string(home.join(LOG)).unwrap();
377	1×	assert!(log.contains("rolled back") && log.contains("0.3.2"), "{log}");
378	1×	}
379
380		#[test]
381	1×	fn only_a_running_runner_is_told() {
382	1×	assert!(is_runner(b"/usr/local/bin/flow\0runner\0"));
383	1×	assert!(is_runner(b"flow\0--home\0/srv/flow\0runner\0"));
384	1×	assert!(!is_runner(b"flow\0runner\0update\0--from\0x\0"), "not the update command itself");
385	1×	assert!(!is_runner(b"/usr/bin/vim\0runner\0"));
386	1×	assert!(!is_runner(b""));
387	1×	let dir = tempfile::tempdir().unwrap();
388	1×	std::fs::write(dir.path().join(PID), std::process::id().to_string()).unwrap();
389	1×	assert_eq!(running(dir.path()), None, "this test is not a flow runner");
390	1×	}
391
392		#[test]
393	1×	fn a_proven_build_stays() {
394	1×	let dir = tempfile::tempdir().unwrap();
395	1×	let exe = dir.path().join("flow");
396	1×	std::fs::write(&exe, "new").unwrap();
397	1×	let p = Pending { from: "0.3.1".into(), to: "0.3.2".into(), exe: exe.clone(), at_ms: 1, notice_secs: 0, starts: 1 };
398	1×	save(dir.path(), &p).unwrap();
399	1×	proven(dir.path());
400	1×	assert!(pending(dir.path()).is_none());
401	1×	assert_eq!(on_start(dir.path()).unwrap(), Start::Settled);
402	1×	assert!(std::fs::read_to_string(dir.path().join(LOG)).unwrap().contains("proven"));
403	1×	}
404		}

Coverage / crates/flow/src/update.rs 59.5%

Keyboard