A runner declines a job it cannot run b11e30f5031d

- **A runner declines a job it cannot run (#36, spec §11.5).** Before fetching anything it checks the tier, the architecture, that the image is there and verifies, and that the sandbox's memory is free. If any check fails it hands the job back to the hub with the reason instead of running and failing it.

/// `POST /api/v1/flow/jobs/{id}/decline` (spec §11.5): the runner will

/// not run a job it was handed, and says why. The hub queues the job

/// again and offers it to this runner no more until its profile changes.

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

pub struct Decline {

    pub reason: String,

}

use flow_core::api::{Claim, ClaimRequest, Decline, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};

    /// Hand job `job` back, saying why this runner will not run it.

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

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

        Ok(())

    }

}

/// Why this runner cannot run `claim` now (spec §11.5), checked before

/// anything is fetched: the tier, the architecture, the image and the

/// memory. The hub chose it from what the runner last said, which may be

/// out of date.

fn preflight(home: &Home, claim: &Claim) -> Option<String> {

    let sb = &claim.sandbox;

    match sb.runtime {

        flow_core::Runtime::Microvm => {

            if let Err(e) = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm") {

                return Some(format!("the MicroVM tier needs /dev/kvm, which cannot be opened here ({e})"));

            }

            if home.config.microvm.is_none() {

                return Some("the MicroVM tier is not set up here (flow.toml has no [microvm] section)".into());

            }

        }

        flow_core::Runtime::Syscall => {

            if let Err(e) = crate::syscall::available() {

                return Some(format!("the syscall tier cannot run here: {e}"));

            }

        }

        flow_core::Runtime::Wasm => return Some("this runner has no wasm tier".into()),

    }

    sandbox_problem(home, sb, std::env::consts::ARCH, available_memory_mib())

}

/// What stops `sb` running on this home's images, on a machine of

/// architecture `here` with `free_mib` of memory free.

fn sandbox_problem(home: &Home, sb: &flow_core::Sandbox, here: &str, free_mib: Option<u64>) -> Option<String> {

    if let Some(want) = sb.arch.as_deref().filter(|a| *a != here) {

        return Some(format!("the sandbox needs {want}; this runner is {here}"));

    }

    let short = &sb.rootfs[..12.min(sb.rootfs.len())];

    let image = home.rootfs(&sb.rootfs);

    if !image.is_file() {

        return Some(format!("rootfs {short} is not on this runner"));

    }

    if let Some(arch) = home.image_arch(&sb.rootfs).filter(|a| a != here) {

        return Some(format!("rootfs {short} is for {arch}; this runner is {here}"));

    }

    if let Err(e) = home.verified(&image, &sb.rootfs) {

        return Some(format!("rootfs {short} does not verify here: {e:#}"));

    }

    if let Some(free) = free_mib.filter(|free| *free < u64::from(sb.memory_mib)) {

        return Some(format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));

    }

    None

}

/// `MemAvailable` from /proc/meminfo, in MiB.

fn available_memory_mib() -> Option<u64> {

    let info = std::fs::read_to_string("/proc/meminfo").ok()?;

    let kib: u64 = info.lines().find_map(|l| l.strip_prefix("MemAvailable:"))?.trim().trim_end_matches("kB").trim().parse().ok()?;

    Some(kib / 1024)

                        if let Some(why) = preflight(home, &claim) {

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

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

                                eprintln!("job {id}: decline: {e:#}");

                            }

                            held.lock().unwrap().remove(&id);

                            let mut u = used.lock().unwrap();

                            u.jobs -= 1;

                            u.memory_mib -= mem;

                            u.cpus -= cpus;

                            return;

                        }

    #[test]

    fn a_runner_names_what_stops_a_sandbox() {

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

        let home = Home::locate(Some(dir.path())).unwrap();

        let bytes = b"not really ext4";

        let hash = blake3::hash(bytes).to_hex().to_string();

        let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({

            "name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"

        }))

        .unwrap();

        let problem = |sb: &flow_core::Sandbox, free| sandbox_problem(&home, sb, "x86_64", free);

        assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");

        std::fs::write(home.rootfs(&hash), bytes).unwrap();

        sb.rootfs = hash.clone();

        assert_eq!(problem(&sb, Some(4096)), None);

        assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");

        std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();

        assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));

        sb.arch = Some("aarch64".into());

        assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");

        sb.arch = None;

        std::fs::remove_file(home.image_manifest(&hash)).unwrap();

        std::fs::write(home.rootfs(&hash), b"changed").unwrap();

        assert!(problem(&sb, None).unwrap().contains("does not verify"));

    }

//! busybox is at /usr/bin/busybox (or `FLOW_TEST_BUSYBOX`); skipped

//! otherwise.

    let busybox = std::env::var_os("FLOW_TEST_BUSYBOX").map(std::path::PathBuf::from).unwrap_or_else(|| "/usr/bin/busybox".into());

    let busybox = busybox.as_path();