flow builds for Windows, offering the WASM tier 3554afefb6aa

## Unreleased

- **The syscall tier's loopback reaches TCP (#73):** with `network "loopback"`, a job may open IPv4 and IPv6 sockets, which in its own network namespace reach only its `lo`, so a server it starts on 127.0.0.1 answers it (a conformance suite over an emulator's HTTP control port, which skipped every test on narvi-1). With `network "isolated"` sockets other than `AF_UNIX` are still refused.

- **Runner builds in releases (oxygenhub#322):** a `dist` job (`scripts/dist.sh`, on a change to `Cargo.toml` or the script) builds `flow` stripped for Linux x86_64, aarch64 and riscv64 (glibc 2.28 and later) as `flow-linux-ARCH`, named by the architecture a runner reports, with `SHA256SUMS`: the files a release carries for runners that update themselves.

- **Runners ask the hub for their update (oxygenhub#322):** once the hub has answered a claim, and every 15 minutes, a runner asks `GET /api/v1/flow/runner/update`; when the hub offers a build, it downloads it (kept in the home's `updates/`), refuses it unless its size and BLAKE3 are those the hub attests, and installs it as `flow runner update` does, leaving within 600 s, or for a dry run checks it and says what would happen. What it did goes back to the hub once per outcome, and a proof or a rollback is told at the next start. `[runner.update] allow = false` refuses and says so; a hub without runner updates (404) is simply asked again later.

//! only if an older lockfile comes back, and then cargo rebuilds it.

use serde::{Deserialize, Serialize};

use std::collections::{HashMap, HashSet};

use std::os::unix::fs::MetadataExt;

use std::time::{Duration, SystemTime};

/// The share of a cache disk a kept cache may fill when no runner says

    t

}

/// Whether the file system holding `path` records reads (not `noatime`).

pub fn records_reads(path: &Path) -> bool {

    let Ok(c) = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()) else { return false };

    st.f_flag & libc::ST_NOATIME == 0

}

/// The blocks in use on the file system mounted at `path`, and its size

/// (as `df` counts them).

pub fn usage(path: &Path) -> std::io::Result<(u64, u64)> {

    Ok(((st.f_blocks as u64).saturating_sub(st.f_bfree as u64) * unit, st.f_blocks as u64 * unit))

}

/// Set every regular file's atime under `root` to the epoch, its mtime

/// kept, so the job's first read of each is recorded even on `relatime`

/// (which records a read only of a file read no later than it changed).

    }

}

/// Each entry under `root` (not `root` itself) with its metadata, parents

/// before children; links are not followed, nor other file systems

/// entered.

    Ok(())

}

/// The unit hashes in `rel`: a 16-digit lowercase hex word after a `-`

/// in any of its names, as cargo names a unit's files.

fn hashes(rel: &Path) -> impl Iterator<Item = String> + '_ {

    })

}

fn nanos(s: i64, ns: i64) -> i128 {

    s as i128 * 1_000_000_000 + ns as i128

}

/// One file, however many names it has.

struct File {

    paths: Vec<PathBuf>,

    used: bool,

}

/// Prune the cache at `root`, a disk of `size` bytes with `before` in

/// use, down to `percent` of its size, removing only regular files this

/// run did not use (see the module's notes), those used longest ago

    Ok(out)

}

mod tests {

    use super::*;

use anyhow::{Context, Result};

use flow_core::pipeline::Mount;

use flow_init::prune::Pruned;

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

use std::time::SystemTime;

            seal(disk).with_context(|| format!("seal {}", disk.display()))?;

        }

        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));

        let dest = mine.join(format!("{key}.{}", ctx.form.ext()));

        if ctx.form == Form::Dir {

            // Another job of this slot kept one since: this one is newer.

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

        {

            use std::io::Write;

        }

        std::fs::rename(&tmp, sidecar)?;

        // One disk per key is enough; older keys of this bookmark go once

        kept(&self.root, 4, &mut |p, form| {

            if let Ok(md) = std::fs::symlink_metadata(p) {

                let bytes = match form {

                    Form::Dir => tree_bytes(p),

                };

                disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), bytes, p.to_path_buf()));

fn tree_bytes(dir: &Path) -> u64 {

    fn walk(p: &Path, seen: &mut std::collections::HashSet<(u64, u64)>) -> u64 {

        let Ok(md) = std::fs::symlink_metadata(p) else { return 0 };

            return 0;

        }

        if md.is_dir() {

            for e in std::fs::read_dir(p).into_iter().flatten().flatten() {

                bytes += walk(&e.path(), seen);

    if md.file_type().is_symlink() {

        return Ok(());

    }

    let want = (mode & !0o7022) | if md.is_dir() { 0o700 } else { 0o400 };

    if want != mode {

    }

    if md.is_dir() {

        for e in std::fs::read_dir(p)? {

    Ok(())

}

mod tests {

    use super::*;

    use flow_core::pipeline::MountSource;

    /// Pruning that never prunes: the cap is the whole disk.

    fn keep_all() -> Prune<'static> {

}

fn entry(out: &mut Vec<u8>, kind: u8, path: &str, meta: &std::fs::Metadata, bytes: &[u8]) {

    let since = meta.modified().ok().and_then(|t| t.duration_since(UNIX_EPOCH).ok()).unwrap_or_default();

    out.push(kind);

    out.extend((path.len() as u32).to_le_bytes());

    out.extend(path.as_bytes());

    out.extend(since.as_secs().to_le_bytes());

    out.extend(since.subsec_nanos().to_le_bytes());

    out.extend((bytes.len() as u64).to_le_bytes());

/// Restore a checkpoint: each entry at `place(checkpoint)` joined with the

/// rest of its path, for the job's own `checkpoints` only.

pub fn unpack(bytes: &[u8], checkpoints: &[String], place: impl Fn(&str) -> Option<PathBuf>) -> Result<Counts> {

    let Some(mut rest) = bytes.strip_prefix(MAGIC) else { bail!("not a checkpoint") };

    let mut n = Counts::default();

    let take = |rest: &mut &[u8], k: usize| -> Result<Vec<u8>> {

                }

                std::fs::write(&at, &data)?;

                let f = std::fs::File::options().write(true).open(&at)?;

                f.set_modified(SystemTime::UNIX_EPOCH + Duration::new(secs, nanos))?;

                n.files += 1;

                n.bytes += data.len() as u64;

//! decision goes into the job log, and the bytes are relayed untouched (no

//! TLS is opened here).

use std::io::{Read, Write};

use std::os::unix::net::{UnixListener, UnixStream};

use std::path::Path;

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

use std::time::Duration;

/// The vsock port the guest's relay connects to on the host (CID 2).

pub const GUEST_PROXY: &str = "http://127.0.0.1:3128";

/// The only port relayed: TLS.

const PORT: u16 = 443;

/// Connections at once, per job.

const MAX_OPEN: usize = 64;

const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// The longest request head read.

const MAX_HEAD: usize = 8 * 1024;

    }

}

/// The proxy for one job: listening until dropped.

pub struct Proxy {

    stop: Arc<AtomicBool>,

    path: std::path::PathBuf,

}

impl Drop for Proxy {

    fn drop(&mut self) {

        self.stop.store(true, Ordering::SeqCst);

    }

}

/// Listen on `socket` (Firecracker's `uds_path` and `_3128`) for the

/// guest's connections; each decision is sent to `log` as a line.

pub fn start(socket: &Path, allow: Vec<String>, log: mpsc::Sender<String>) -> std::io::Result<Proxy> {

    Ok(Proxy { stop, path: socket.to_path_buf() })

}

fn refuse(mut conn: UnixStream, why: &str) -> std::io::Result<()> {

    let body = format!("{why}\n");

    write!(conn, "HTTP/1.1 403 Forbidden\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len())

}

/// Read up to the blank line that ends a request head.

fn read_head(conn: &mut UnixStream) -> std::io::Result<String> {

    let mut head = Vec::new();

    Ok(String::from_utf8_lossy(&head).into_owned())

}

fn serve(mut conn: UnixStream, allow: &[String], log: &mpsc::Sender<String>) {

    let _ = conn.set_read_timeout(Some(Duration::from_secs(30)));

    let head = match read_head(&mut conn) {

    relay(conn, out);

}

/// Copy both ways until either side closes.

fn relay(guest: UnixStream, out: TcpStream) {

    let (Ok(mut guest_r), Ok(mut out_w)) = (guest.try_clone(), out.try_clone()) else { return };

    let _ = up.join();

}

mod tests {

    use super::*;

/// The kernel's release, as uname says. Not read from /proc: an Android

/// app may not read it there, and is refused (and logged) every claim.

pub fn kernel() -> Option<String> {

    let mut u: libc::utsname = unsafe { std::mem::zeroed() };

    // SAFETY: `u` is a writable utsname.

    Some(release.to_string_lossy().into_owned())

}

pub fn free_disk_mib(path: &std::path::Path) -> Option<u64> {

    use std::os::unix::ffi::OsStrExt;

    let c = std::ffi::CString::new(path.as_os_str().as_bytes()).ok()?;

        return None;

    }

    Some((s.f_bavail as u64).saturating_mul(s.f_frsize as u64) >> 20)

}

/// The disk the files under `dir` take (their blocks, not their sizes: a

/// job's disks are sparse), in MiB; links are not followed.

pub fn allocated_mib(dir: &std::path::Path) -> Option<u64> {

    let mut total: u64 = 0;

    let mut stack = vec![dir.to_path_buf()];

    while let Some(d) = stack.pop() {

            if m.is_dir() {

                stack.push(e.path());

            } else {

            }

        }

    }

}

/// The 1-minute load average, times 100, from sysinfo (the kernel's own

pub fn load_x100() -> Option<u32> {

    let mut i: libc::sysinfo = unsafe { std::mem::zeroed() };

    // SAFETY: `i` is a writable sysinfo.

        let tmp = self.root.join("runner.key.tmp");

        {

            use std::io::Write;

            f.write_all(&secret.encode(&identity.hash()))?;

            f.sync_all()?;

        }

pub mod facts;

/// The job-starting code `flow-init` uses in a microVM, shared with the

/// syscall tier.

#[path = "../../flow-init/src/job.rs"]

pub mod guest;

pub mod home;

pub mod metrics;

pub mod microvm;

pub mod outputs;

pub mod runner;

pub mod swarm;

pub mod syscall;

pub mod update;

pub mod wait;

pub mod wasm;

            // 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)? {

            }

            // The runner's oxy state (identity, user config) stays in its home.

            if std::env::var_os("OXY_HOME").is_none() {

/// [`MAX_ARTIFACT_BYTES`]. The job wrote that tree, so no link is

/// followed on the way: each directory is opened with `O_NOFOLLOW`, and

/// so is the file.

pub fn read_dir_file(base: &Path, inner: &str) -> Result<Vec<u8>> {

    use std::ffi::CString;

    use std::os::fd::FromRawFd;

    let c = |s: &str| CString::new(s).context("a path holds NUL");

    let mut dir = unsafe { libc::open(c(&base.to_string_lossy())?.as_ptr(), libc::O_DIRECTORY | libc::O_RDONLY | libc::O_CLOEXEC) };

    if dir < 0 {

    unsafe { libc::close(dir) };

    if fd < 0 {

        bail!("{inner} is not a regular file in the job's files ({err})");

    }

    let meta = file.metadata()?;

    if !meta.is_file() {

        bail!("{inner} is not a regular file in the job's files (it may be missing or a link)");

    let path = home.root.join(crate::home::RUNNER_TOML);

    {

        use std::io::Write;

        let tmp = path.with_extension("toml.tmp");

        f.write_all(text.as_bytes())?;

        f.sync_all()?;

        std::fs::rename(&tmp, &path)?;

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

    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();

/// 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 made = rcgen::generate_simple_self_signed(vec!["flow-swarm".to_string()]).context("make the swarm's certificate")?;

    let (cert, key) = (made.cert.der().to_vec(), made.signing_key.serialize_der());

    {

        f.write_all(&key)?;

    }

    std::fs::write(&cert_path, &cert)?;

use crate::guest::{dev_links, job_env, loopback_up, resolve_program, PATH};

use crate::home::Home;

use crate::microvm::{Outcome, Sink};

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

use flow_core::api::{Record, SecretInput};

use flow_core::{Environment, Job, MountSource, Network, Sandbox};

use serde::{Deserialize, Serialize};

use std::ffi::CString;

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

use std::sync::{mpsc, OnceLock};

use std::time::{Duration, Instant};

    }

}

fn kernel_release() -> String {

    crate::facts::kernel().unwrap_or_else(|| "unknown".into())

}

        peak_memory_mib,

        pruned: Vec::new(),

    })

}

/// The image's tree, unpacked from its ext4 file once per hash with

/// 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");

    }

    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()))?;

    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)");

        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");

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

}

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

                }

                if *kind == NodeKind::Symlink {

                    let to = String::from_utf8_lossy(&bytes).into_owned();

                } else {

                    std::fs::write(&target, &bytes).with_context(|| format!("write {path}"))?;

                    if f.mode.exec {

                    }

                }

                files.push((path.clone(), *blake3::hash(&bytes).as_bytes()));

            std::fs::create_dir_all(&target)?;

        } else if ft.is_symlink() {

            let to = std::fs::read_link(entry.path())?;

            files.push((rel_s, *blake3::hash(to.to_string_lossy().as_bytes()).as_bytes()));

        } else if ft.is_file() {

            std::fs::copy(entry.path(), &target)?;

        .map(|m| m.len().max(4096))

        .sum();

    let mib = used * 12 / 10 / (1 << 20) + 512;

    let status = std::process::Command::new("mkfs.ext4")

        .args(["-q", "-F", "-L", "workspace", "-E"])

        .arg(format!("root_owner={uid}:{gid}"))

/// the workspace. mkfs.ext4 creates the file sparse, so only what the job

/// writes takes space on the runner.

pub fn empty_ext4(out: &Path, mib: u32, label: &str) -> Result<()> {

    let status = std::process::Command::new("mkfs.ext4")

        .args(["-q", "-F", "-L", label, "-E"])

        .arg(format!("root_owner={uid}:{gid}"))

    }

}

fn open_up(dir: &Path) {

    let Ok(meta) = std::fs::symlink_metadata(dir) else { return };

    if !meta.is_dir() {

        return;

    }

    if let Ok(entries) = std::fs::read_dir(dir) {

        for e in entries.flatten() {

            open_up(&e.path());