Ephemeral scratch mounts in M0: ephemeral://tmpfs and ephemeral://disk (a sparse disk per job), checked targets and sizes, in the canonical job and the receipt; flow-init mounts them for the job's user; make deploy-runner fc10d9e80383

# OxygenFlow on Earth (docs/m0.md §6): build the runner and the guest init,

# install them under /home/claude/flow, re-pin flow-init in flow.toml, and

# restart the flow-runner service.

EARTH ?= claude@earth

SSH ?= ssh -i $(HOME)/.ssh/earth_claude

SCP ?= scp -i $(HOME)/.ssh/earth_claude

FLOW_HOME_ON_EARTH ?= /home/claude/flow

INIT := target/x86_64-unknown-linux-gnu/release/flow-init

.PHONY: build deploy-runner

build:

	cargo build --release -p flow

	RUSTFLAGS="-C target-feature=+crt-static" cargo build --release -p flow-init --target x86_64-unknown-linux-gnu

deploy-runner: build

	$(SCP) target/release/flow $(EARTH):$(FLOW_HOME_ON_EARTH)/bin/flow.new

	$(SCP) $(INIT) $(EARTH):$(FLOW_HOME_ON_EARTH)/bin/flow-init.new

	$(SSH) $(EARTH) 'set -e; cd $(FLOW_HOME_ON_EARTH); \

		cp -p bin/flow bin/flow.prev; cp -p bin/flow-init bin/flow-init.prev; cp -p flow.toml flow.toml.prev; \

		mv bin/flow.new bin/flow; mv bin/flow-init.new bin/flow-init; \

		h=$$(bin/flow hash bin/flow-init | cut -d" " -f1); \

		sed -i "s|^init = .*|init = { path = \"bin/flow-init\", blake3 = \"$$h\" }|" flow.toml; \

		sudo systemctl restart flow-runner; systemctl is-active flow-runner'

## Deploying the runner on Earth

`make deploy-runner` builds `flow` and the static `flow-init`, installs them

under `/home/claude/flow` (keeping the previous files as `.prev`), re-pins

`flow-init` in `flow.toml` and restarts the `flow-runner` service.

    let mut fields = vec![

    ];

    // Mounts as [source, target, size MiB] sorted by target; absent when

    // there are none, so jobs without mounts hash as before.

    if !j.mounts.is_empty() {

        fields.push((

            6,

            Value::Array(

                j.mounts

                    .iter()

                    .map(|m| Value::Array(vec![text(m.source.as_str()), text(&m.target), Value::Uint(m.size_mib as u64)]))

                    .collect(),

            ),

        ));

    }

    Value::map(fields)

pub use pipeline::{Job, Mount, MountSource, Network, Pipeline, Problem, Problems, Runtime, Sandbox, PIPELINE_PATH};

    /// Ephemeral scratch mounts, sorted by target (spec §3.1 `mount`).

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub mounts: Vec<Mount>,

}

/// An ephemeral scratch mount: empty at the start of every job and gone

/// after it, so it cannot carry anything between jobs (spec §2.2 I4).

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

pub struct Mount {

    pub source: MountSource,

    /// An absolute path under `/src` or `/tmp`.

    pub target: String,

    pub size_mib: u32,

}

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

#[serde(rename_all = "lowercase")]

pub enum MountSource {

    /// `ephemeral://tmpfs`: memory in the sandbox, counted against its memory-limit.

    Tmpfs,

    /// `ephemeral://disk`: a fresh sparse disk on the runner, for large build trees.

    Disk,

}

impl MountSource {

    pub fn as_str(self) -> &'static str {

        match self {

            MountSource::Tmpfs => "ephemeral://tmpfs",

            MountSource::Disk => "ephemeral://disk",

        }

    }

}

/// Mounts a job may declare.

pub const MAX_MOUNTS: usize = 4;

/// Sizes a mount may have: 16 MiB to 64 GiB.

pub const MIN_MOUNT_MIB: u64 = 16;

pub const MAX_MOUNT_MIB: u64 = 64 * 1024;

/// Whether `t` is a clean absolute path strictly under `/src` or `/tmp`.

fn mount_target_ok(t: &str) -> bool {

    let under = t.strip_prefix("/src/").or_else(|| t.strip_prefix("/tmp/"));

    under.is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != ".."))

const LATER_JOB: &[&str] = &["output", "secret", "attest"];

/// `mount { source "ephemeral://disk"; target "/src/target"; size "8GiB" }`.

fn parse_mount(ctx: &mut Ctx, node: &KdlNode) -> Option<Mount> {

    let (mut source, mut target, mut size) = (None, None, None);

    let mut seen = BTreeSet::new();

    for c in block(ctx, node) {

        let key = c.name().value();

        if !seen.insert(key.to_string()) {

            ctx.err(c.span(), format!("`{key}` is given twice"));

            continue;

        }

        ctx.no_children(c);

        match key {

            "source" => {

                source = match ctx.one_string(c).as_deref() {

                    Some("ephemeral://tmpfs") => Some(MountSource::Tmpfs),

                    Some("ephemeral://disk") => Some(MountSource::Disk),

                    Some(other) => {

                        ctx.err(

                            c.span(),

                            format!("mount source {other:?} is not in M0 yet; M0 mounts \"ephemeral://tmpfs\" or \"ephemeral://disk\""),

                        );

                        None

                    }

                    None => None,

                }

            }

            "target" => {

                target = match ctx.one_string(c) {

                    Some(t) if mount_target_ok(&t) => Some(t),

                    Some(t) => {

                        ctx.err(

                            c.span(),

                            format!("mount target {t:?} must be an absolute path under /src or /tmp, without . or .. or a trailing /"),

                        );

                        None

                    }

                    None => None,

                }

            }

            "size" => {

                size = match ctx.one_string(c) {

                    Some(m) => match parse_memory(&m) {

                        Some(mib) if (MIN_MOUNT_MIB..=MAX_MOUNT_MIB).contains(&mib) => Some(mib as u32),

                        Some(_) => {

                            ctx.err(c.span(), format!("mount size {m:?} must be from 16MiB to 64GiB"));

                            None

                        }

                        None => {

                            ctx.err(c.span(), format!("mount size {m:?}: expected a whole number of KiB, MiB or GiB, as \"8GiB\""));

                            None

                        }

                    },

                    None => None,

                }

            }

            "readonly" | "include" => ctx.err(c.span(), format!("mount `{key}` is not in M0 yet")),

            other => ctx.err(c.span(), format!("unknown mount setting `{other}`")),

        }

    }

    for (name, missing) in [("source", source.is_none()), ("target", target.is_none()), ("size", size.is_none())] {

        if missing && !seen.contains(name) {

            ctx.err(node.span(), format!("a mount needs a {name}"));

        }

    }

    Some(Mount { source: source?, target: target?, size_mib: size? })

}

    let mut job = Job {

        name: name.clone(),

        sandbox: String::new(),

        scope: vec![],

        depends_on: vec![],

        env: BTreeMap::new(),

        exec: vec![],

        mounts: vec![],

    };

        if key != "scope" && key != "depends-on" && key != "mount" && !seen.insert(key.to_string()) {

            "mount" => {

                if !c.entries().is_empty() {

                    ctx.err(c.span(), "`mount` takes a block of source, target and size");

                }

                if let Some(m) = parse_mount(ctx, c) {

                    if job.mounts.len() == MAX_MOUNTS {

                        ctx.err(c.span(), format!("a job has at most {MAX_MOUNTS} mounts"));

                    } else if let Some(other) = job.mounts.iter().find(|o| {

                        o.target == m.target

                            || m.target.starts_with(&format!("{}/", o.target))

                            || o.target.starts_with(&format!("{}/", m.target))

                    }) {

                        ctx.err(c.span(), format!("mount {:?} overlaps mount {:?}", m.target, other.target));

                    } else {

                        job.mounts.push(m);

                    }

                }

            }

    job.mounts.sort_by(|a, b| a.target.cmp(&b.target));

    assert!(has(14, "mount source \"oxy://workspace\" is not in M0 yet"), "{ps:?}");

}

#[test]

fn ephemeral_mounts_parse_sort_and_hash() {

    let plain = Pipeline::parse(&doc("job \"b\" {\n    sandbox \"alpine\"\n    exec \"cargo\" \"build\"\n}\n")).unwrap();

    let p = Pipeline::parse(&doc(r#"job "b" {

    sandbox "alpine"

    mount {

        source "ephemeral://tmpfs"

        target "/tmp/cache"

        size "256MiB"

    }

    mount {

        source "ephemeral://disk"

        target "/src/target"

        size "8GiB"

    }

    exec "cargo" "build"

}

"#))

    .unwrap();

    let b = &p.jobs["b"];

    assert_eq!(

        b.mounts,

        vec![

            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192 },

            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256 },

        ],

        "sorted by target"

    );

    assert_ne!(p.hash(), plain.hash(), "mounts are part of the job");

    // A job without mounts encodes as before mounts existed: no field 6.

    let v = oxy_core::cbor::decode(&plain.canonical()).unwrap();

    let text = format!("{v:?}");

    assert!(!text.contains("ephemeral"), "{text}");

    let json = serde_json::to_value(&plain.jobs["b"]).unwrap();

    assert!(json.get("mounts").is_none(), "absent from the API too");

    let back: Job = serde_json::from_value(json).unwrap();

    assert!(back.mounts.is_empty());

}

#[test]

fn mounts_are_checked() {

    let job = |mount: &str| doc(&format!("job \"b\" {{\n    sandbox \"alpine\"\n    mount {{\n{mount}\n    }}\n    exec \"true\"\n}}\n"));

    let has = |src: String, want: &str| {

        let ps = problems(&src);

        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{want:?} not in {ps:?}");

    };

    has(job("source \"oxy://workspace\"\ntarget \"/src/x\"\nsize \"1GiB\""), "not in M0 yet");

    has(job("source \"ephemeral://disk\"\ntarget \"/etc/x\"\nsize \"1GiB\""), "under /src or /tmp");

    has(job("source \"ephemeral://disk\"\ntarget \"/src\"\nsize \"1GiB\""), "under /src or /tmp");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/../etc\"\nsize \"1GiB\""), "without . or ..");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/x/\"\nsize \"1GiB\""), "trailing /");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"8MiB\""), "from 16MiB to 64GiB");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"65GiB\""), "from 16MiB to 64GiB");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"lots\""), "whole number");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\""), "needs a size");

    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"1GiB\"\nreadonly #true"), "`readonly` is not in M0 yet");

    has(job("source \"ephemeral://disk\"\nsource \"ephemeral://tmpfs\"\ntarget \"/src/x\"\nsize \"1GiB\""), "given twice");

    let two = |a: &str, b: &str| {

        doc(&format!(

            "job \"b\" {{\n    sandbox \"alpine\"\n    mount {{ source \"ephemeral://disk\"; target \"{a}\"; size \"1GiB\"; }}\n    mount {{ source \"ephemeral://disk\"; target \"{b}\"; size \"1GiB\"; }}\n    exec \"true\"\n}}\n"

        ))

    };

    has(two("/src/t", "/src/t/deps"), "overlaps");

    has(two("/src/t", "/src/t"), "overlaps");

    assert!(Pipeline::parse(&two("/src/t", "/src/tt")).is_ok(), "a shared prefix is not nesting");

    let five: String =

        (0..5).map(|i| format!("    mount {{ source \"ephemeral://tmpfs\"; target \"/tmp/m{i}\"; size \"16MiB\"; }}\n")).collect();

    has(doc(&format!("job \"b\" {{\n    sandbox \"alpine\"\n{five}    exec \"true\"\n}}\n")), "at most 4 mounts");

    #[serde(default)]

    mounts: Vec<MountSpec>,

}

/// An ephemeral mount from the host: a tmpfs of `size_mib`, or the disk at

/// `device`, mounted at `target` under the new root.

#[derive(serde::Deserialize)]

struct MountSpec {

    kind: String,

    target: String,

    size_mib: u32,

    #[serde(default)]

    device: Option<String>,

fn chown(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {

    let p = cstr(path)?;

    if unsafe { libc::chown(p.as_ptr(), uid, gid) } != 0 {

        return Err(format!("chown {path}: {}", std::io::Error::last_os_error()));

    }

    Ok(())

}

/// Create `path` and its missing parents, each owned by the job's user.

fn make_dirs_owned(path: &str, uid: libc::uid_t, gid: libc::gid_t) -> Result<(), String> {

    let mut missing = Vec::new();

    let mut p = std::path::Path::new(path);

    while !p.exists() {

        missing.push(p.to_path_buf());

        p = match p.parent() {

            Some(parent) => parent,

            None => break,

        };

    }

    for d in missing.iter().rev() {

        std::fs::create_dir(d).map_err(|e| format!("mkdir {}: {e}", d.display()))?;

        chown(&d.to_string_lossy(), uid, gid)?;

    }

    Ok(())

}

    // Ephemeral mounts: mount points are made (and owned by the job's user)

    // under the writable /src or /tmp, then each is mounted and given to it.

    for m in &job.mounts {

        let target = format!("/newroot{}", m.target);

        make_dirs_owned(&target, uid, gid)?;

        match (m.kind.as_str(), &m.device) {

            ("tmpfs", _) => mount("tmpfs", &target, "tmpfs", nosuid, &format!("size={}m,uid={uid},gid={gid},mode=755", m.size_mib))?,

            ("disk", Some(dev)) => {

                mount(dev, &target, "ext4", nosuid, "")?;

                chown(&target, uid, gid)?;

            }

            (kind, _) => return Err(format!("unknown mount kind {kind}")),

        }

    }

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

        ]

        .into_iter()

        .chain(job.mounts.iter().map(|m| (format!("mount {}", m.target), format!("{} {} MiB", m.source.as_str(), m.size_mib))))

        .collect(),

    let spec = guest_spec(job);

    // Each `ephemeral://disk` mount is a fresh sparse ext4 disk in the

    // scratch directory, attached after the workspace (vdc, vdd, ...).

    let mut disks = Vec::new();

    for (i, m) in job.mounts.iter().filter(|m| m.source == MountSource::Disk).enumerate() {

        let path = scratch.join(format!("mount-{i}.ext4"));

        crate::workspace::empty_ext4(&path, m.size_mib, "scratch")?;

        disks.push(

            serde_json::json!({ "drive_id": format!("mount{i}"), "path_on_host": path, "is_root_device": false, "is_read_only": false }),

        );

    }

    let drives: Vec<serde_json::Value> = [

        serde_json::json!({ "drive_id": "rootfs", "path_on_host": rootfs, "is_root_device": false, "is_read_only": true }),

        serde_json::json!({ "drive_id": "workspace", "path_on_host": workspace, "is_root_device": false, "is_read_only": false }),

    ]

    .into_iter()

    .chain(disks)

    .collect();

        "drives": drives,

/// What `flow-init` reads from `/job.json`: argv, env, and the mounts

/// with the device each disk mount is attached as (vdc onwards).

pub fn guest_spec(job: &Job) -> serde_json::Value {

    let mut next_disk = b'c';

    let mounts: Vec<serde_json::Value> = job

        .mounts

        .iter()

        .map(|m| match m.source {

            MountSource::Tmpfs => serde_json::json!({ "kind": "tmpfs", "target": m.target, "size_mib": m.size_mib }),

            MountSource::Disk => {

                let dev = format!("/dev/vd{}", next_disk as char);

                next_disk += 1;

                serde_json::json!({ "kind": "disk", "target": m.target, "size_mib": m.size_mib, "device": dev })

            }

        })

        .collect();

    serde_json::json!({ "argv": job.exec, "env": job.env.iter().collect::<Vec<_>>(), "mounts": mounts })

}

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    use flow_core::Mount;

    #[test]

    fn guest_spec_numbers_disks_after_the_workspace() {

        let mut job: Job = serde_json::from_value(serde_json::json!({

            "name": "t", "sandbox": "s", "scope": [], "depends_on": [], "env": {"A": "1"}, "exec": ["true"]

        }))

        .unwrap();

        assert_eq!(guest_spec(&job)["mounts"], serde_json::json!([]), "no mounts");

        job.mounts = vec![

            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192 },

            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256 },

            Mount { source: MountSource::Disk, target: "/tmp/z".into(), size_mib: 64 },

        ];

        let spec = guest_spec(&job);

        assert_eq!(spec["mounts"][0], serde_json::json!({"kind": "disk", "target": "/src/target", "size_mib": 8192, "device": "/dev/vdc"}));

        assert_eq!(spec["mounts"][1], serde_json::json!({"kind": "tmpfs", "target": "/tmp/cache", "size_mib": 256}));

        assert_eq!(spec["mounts"][2]["device"], "/dev/vdd");

        assert_eq!(spec["env"], serde_json::json!([["A", "1"]]));

/// An empty ext4 filesystem of `mib` MiB at `out`, owned by this user like

/// 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 (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) };

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

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

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

        .arg(out)

        .arg(format!("{mib}M"))

        .status()

        .context("run mkfs.ext4 (e2fsprogs 1.43 or later)")?;

    if !status.success() {

        bail!("mkfs.ext4 failed for a scratch disk");

    }

    Ok(())

}

- `output`, `publish` and `secret` are recognised and refused with

- `mount` is accepted for **ephemeral scratch** only (up to 4 per job):

  `mount { source "ephemeral://disk"; target "/src/target"; size "8GiB" }`.

  `ephemeral://tmpfs` is memory in the guest and counts against its

  memory-limit; `ephemeral://disk` is a fresh sparse ext4 disk on the

  runner, attached after the workspace, for build trees larger than

  memory. The target is a clean absolute path strictly under `/src` or

  `/tmp`, mounts may not nest, sizes run from 16 MiB to 64 GiB, and any

  other source (`oxy://workspace` and the rest) or setting (`readonly`,

  `include`) is refused as not in M0. Both are empty when the job starts

  and deleted when it ends, so nothing passes between jobs. Mounts are

  part of the job's canonical form (field 6, present only when a job has

  mounts, so pipelines without them keep their hashes) and of the

  receipt's environment.

   `/newroot/src`, and tmpfs at `/newroot/tmp` and `/newroot/run`, then the

   job's scratch mounts (creating their mount points, owned by the job's

   user); it

into an image); mounts other than ephemeral scratch, `output` and artifacts; receipt reuse; provenance