Job artifacts: output declarations, extraction and upload a67b97e9a717

//! A job's artifacts (spec §5.2): read from its disks after the microVM

//! has stopped, with debugfs, so the runner needs no mount rights.

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

use flow_core::{Job, MountSource};

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

/// Largest artifact a runner takes: 512 MiB.

pub const MAX_ARTIFACT_BYTES: u64 = 512 << 20;

/// Where guest path `path` lives on the host: the disk image, and the path

/// inside its file system. The workspace is `/src`; each `ephemeral://disk`

/// mount is `mount-N.ext4` in `scratch`, numbered as `microvm::run` does.

pub fn locate(job: &Job, workspace: &Path, scratch: &Path, path: &str) -> (PathBuf, String) {

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

        if let Some(rest) = path.strip_prefix(&format!("{}/", m.target)) {

            return (scratch.join(format!("mount-{i}.ext4")), format!("/{rest}"));

        }

    }

    (workspace.to_path_buf(), path.strip_prefix("/src").unwrap_or(path).to_string())

}

fn debugfs(image: &Path, request: &str) -> Result<String> {

    let out = std::process::Command::new("debugfs").arg("-R").arg(request).arg(image).output().context("run debugfs (e2fsprogs)")?;

    if !out.status.success() {

        bail!("debugfs {request:?}: {}", String::from_utf8_lossy(&out.stderr).trim());

    }

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

}

/// Read regular file `inner` of ext4 image `image`, at most

/// [`MAX_ARTIFACT_BYTES`]. Paths come from a checked pipeline: absolute,

/// clean, and free of quotes and spaces.

pub fn read(image: &Path, inner: &str, scratch: &Path) -> Result<Vec<u8>> {

    let stat = debugfs(image, &format!("stat \"{inner}\""))?;

    if !stat.contains("Type: regular") {

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

    }

    let size: u64 = stat

        .split("Size: ")

        .nth(1)

        .and_then(|s| s.split_whitespace().next())

        .and_then(|s| s.parse().ok())

        .context("debugfs gave no size")?;

    if size > MAX_ARTIFACT_BYTES {

        bail!("{inner} is {} MiB; artifacts are at most {} MiB", size >> 20, MAX_ARTIFACT_BYTES >> 20);

    }

    let dest = scratch.join("artifact.out");

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

    debugfs(image, &format!("dump \"{inner}\" \"{}\"", dest.display()))?;

    let bytes = std::fs::read(&dest).with_context(|| format!("{inner}: debugfs wrote nothing"))?;

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

    if bytes.len() as u64 != size {

        bail!("{inner}: read {} bytes of {size}", bytes.len());

    }

    Ok(bytes)

}

#[cfg(test)]

mod tests {

    use super::*;

    use flow_core::Mount;

    fn job() -> Job {

        Job {

            name: "build".into(),

            sandbox: "rust".into(),

            scope: vec![],

            depends_on: vec![],

            env: Default::default(),

            exec: vec!["make".into()],

            mounts: vec![

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

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

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

            ],

            outputs: vec![],

        }

    }

    #[test]

    fn paths_are_found_on_the_disk_that_holds_them() {

        let (w, s) = (Path::new("/w.ext4"), Path::new("/scratch"));

        assert_eq!(locate(&job(), w, s, "/src/target/release/oxy"), (s.join("mount-0.ext4"), "/release/oxy".into()));

        assert_eq!(locate(&job(), w, s, "/src/work/a"), (s.join("mount-1.ext4"), "/a".into()));

        assert_eq!(locate(&job(), w, s, "/src/docs/a.1"), (w.to_path_buf(), "/docs/a.1".into()));

        assert_eq!(locate(&job(), w, s, "/src/targets/x"), (w.to_path_buf(), "/targets/x".into()), "a prefix is not a parent");

    }

    #[test]

    fn files_are_read_back_from_an_ext4_image() {

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

        let tree = dir.path().join("tree");

        std::fs::create_dir_all(tree.join("release")).unwrap();

        let body: Vec<u8> = (0..200_000u32).map(|i| (i % 251) as u8).collect();

        std::fs::write(tree.join("release/tool"), &body).unwrap();

        std::os::unix::fs::symlink("tool", tree.join("release/link")).unwrap();

        let image = dir.path().join("disk.ext4");

        crate::workspace::image(&tree, &image).unwrap();

        assert_eq!(read(&image, "/release/tool", dir.path()).unwrap(), body);

        assert!(read(&image, "/release/missing", dir.path()).is_err());

        assert!(read(&image, "/release/link", dir.path()).unwrap_err().to_string().contains("not a regular file"));

    }

}

    // Artifacts as [name, path] sorted by name; absent when there are none.

    if !j.outputs.is_empty() {

        fields.push((7, Value::Array(j.outputs.iter().map(|o| Value::Array(vec![text(&o.name), text(&o.path)])).collect())));

    }

    /// Files the job produces, sorted by name (spec §5.2 `output`).

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

    pub outputs: Vec<Output>,

}

/// A file a job produces (spec §5.2): read from the job's disks when it

/// ends, kept in the repository as a blob, and named by its receipt.

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

pub struct Output {

    /// The artifact's name, unique in the job.

    pub name: String,

    /// A clean absolute path under `/src`, not on a tmpfs mount.

    pub path: String,

}

/// Artifacts a job may declare.

pub const MAX_OUTPUTS: usize = 16;

/// Whether `p` is a clean absolute path strictly under `/src`, with nothing

/// a debugfs command line would have to quote.

fn output_path_ok(p: &str) -> bool {

    p.strip_prefix("/src/").is_some_and(|rest| {

        !rest.is_empty()

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

            && p.bytes().all(|b| b.is_ascii_graphic() && b != b'"' && b != b'\\')

    })

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

/// `output { artifact "name" { path "/src/target/release/tool" } }`.

fn parse_output(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {

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

        ctx.err(node.span(), "`output` takes a block of artifacts");

    }

    for a in block(ctx, node) {

        if a.name().value() != "artifact" {

            ctx.err(a.span(), format!("unknown output `{}`; outputs are `artifact \"name\" {{ path \"/src/…\" }}`", a.name().value()));

            continue;

        }

        let Some(name) = ctx.one_string(a) else { continue };

        if !ctx.name_ok(a.span(), "artifact", &name) {

            continue;

        }

        let mut path = None;

        for c in block(ctx, a) {

            ctx.no_children(c);

            match c.name().value() {

                "path" => match ctx.one_string(c) {

                    Some(p) if output_path_ok(&p) => path = Some(p),

                    Some(p) => ctx.err(

                        c.span(),

                        format!("artifact path {p:?} must be an absolute path under /src, without . or .., quotes or spaces"),

                    ),

                    None => {}

                },

                "format" => {

                    if ctx.one_string(c).as_deref() != Some("blob") {

                        ctx.err(c.span(), "artifact format is \"blob\"");

                    }

                }

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

            }

        }

        let Some(path) = path else {

            ctx.err(a.span(), format!("artifact {name:?} needs a path"));

            continue;

        };

        if job.outputs.iter().any(|o| o.name == name) {

            ctx.err(a.span(), format!("artifact {name:?} is given twice"));

        } else if job.outputs.len() == MAX_OUTPUTS {

            ctx.err(a.span(), format!("a job has at most {MAX_OUTPUTS} artifacts"));

        } else {

            job.outputs.push(Output { name, path });

        }

    }

}

        outputs: vec![],

    let mut outputs_at = None;

            "output" => {

                outputs_at = Some(c.span());

                parse_output(ctx, c, &mut job);

            }

    job.outputs.sort_by(|a, b| a.name.cmp(&b.name));

    // A tmpfs is gone when the job ends; nothing on it can be kept.

    for o in &job.outputs {

        if let Some(m) = job.mounts.iter().find(|m| m.source == MountSource::Tmpfs && o.path.starts_with(&format!("{}/", m.target))) {

            ctx.err(

                outputs_at.unwrap_or(node.span()),

                format!(

                    "artifact {:?} is on the tmpfs mount {:?}, which is gone when the job ends; use ephemeral://disk",

                    o.name, m.target

                ),

            );

        }

    }

#[test]

fn artifacts_are_declared_checked_and_hashed() {

    let src = |out: &str| {

        doc(&format!(

            "job \"build\" {{\n    sandbox \"alpine\"\n    mount {{\n        source \"ephemeral://disk\"\n        target \"/src/target\"\n        size \"1GiB\"\n    }}\n    mount {{\n        source \"ephemeral://tmpfs\"\n        target \"/src/tmp\"\n        size \"64MiB\"\n    }}\n    exec \"make\"\n{out}}}\n"

        ))

    };

    let p = Pipeline::parse(&src(

        "    output {\n        artifact \"tool\" {\n            path \"/src/target/release/tool\"\n            format \"blob\"\n        }\n        artifact \"docs\" {\n            path \"/src/docs/tool.1\"\n        }\n    }\n",

    ))

    .unwrap();

    let names: Vec<(&str, &str)> = p.jobs["build"].outputs.iter().map(|o| (o.name.as_str(), o.path.as_str())).collect();

    assert_eq!(names, [("docs", "/src/docs/tool.1"), ("tool", "/src/target/release/tool")], "sorted by name");

    // Artifacts are part of the job's hash; a job without them hashes as before.

    let plain = Pipeline::parse(&src("")).unwrap();

    let sb = &p.sandboxes["alpine"];

    assert_ne!(p.jobs["build"].hash(sb), plain.jobs["build"].hash(sb));

    assert!(plain.jobs["build"].outputs.is_empty());

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

    assert!(!json.contains("outputs"), "no outputs field when there are none: {json}");

    let bad = |out: &str| problems(&src(out)).into_iter().map(|(_, m)| m).collect::<Vec<_>>().join("\n");

    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/etc/passwd\"\n        }\n    }\n")

        .contains("must be an absolute path under /src"));

    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/../etc\"\n        }\n    }\n")

        .contains("must be an absolute path under /src"));

    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/tmp/a\"\n        }\n    }\n").contains("tmpfs"));

    assert!(bad("    output {\n        artifact \"x\" {\n        }\n    }\n").contains("needs a path"));

    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/a\"\n        }\n        artifact \"x\" {\n            path \"/src/b\"\n        }\n    }\n")

        .contains("given twice"));

    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/a\"\n            format \"tar\"\n        }\n    }\n")

        .contains("format"));

}

mod outputs;

    /// Upload artifact `name` of job `job` (spec §5.2); the hub keeps it as

    /// a blob of the job's repository and answers with its ObjectRef.

    pub fn artifact(&self, job: i64, name: &str, bytes: &[u8]) -> Result<String> {

        let url = format!("{}/api/v1/flow/jobs/{job}/artifacts/{name}", self.base);

        let mut delays = RETRY_DELAYS.iter();

        loop {

            let req = self

                .agent

                .post(&url)

                .timeout(Duration::from_secs(600))

                .set("Authorization", &format!("Bearer {}", self.token))

                .set("Content-Type", "application/octet-stream");

            match req.send_bytes(bytes) {

                Ok(r) => {

                    let v: serde_json::Value = r.into_json()?;

                    return v["blob"].as_str().map(String::from).context("the hub named no blob");

                }

                Err(ureq::Error::Status(code, r)) => {

                    let text = r.into_string().unwrap_or_default();

                    match delays.next() {

                        Some(d) if code >= 500 => std::thread::sleep(*d),

                        _ => bail!("{url}: HTTP {code}: {text}"),

                    }

                }

                Err(e) => bail!("{url}: {e}"),

            }

        }

    }

        // A passing job's artifacts, read from its disks and uploaded; a

        // declared artifact that cannot be kept fails the job.

        let mut outputs = vec![];

        let mut output_error = None;

        if outcome.exit_code == Some(0) {

            for o in &claim.job.outputs {

                let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);

                let kept = crate::outputs::read(&image, &inner, &scratch).and_then(|bytes| {

                    let blob = Repo::blob_ref(&bytes);

                    let named = hub.artifact(claim.job_id, &o.name, &bytes)?;

                    if named != blob.to_hex() {

                        bail!("the hub kept it as {named}, not {}", blob.to_hex());

                    }

                    note(&mut sink, format!("artifact {}: {} bytes, blob {}", o.name, bytes.len(), &named[..12]));

                    Ok(blob.0)

                });

                match kept {

                    Ok(b) => outputs.push(b),

                    Err(e) => {

                        output_error = Some(format!("artifact {}: {e:#}", o.name));

                        break;

                    }

                }

            }

        }

        if let Some(e) = &output_error {

            note(&mut sink, e.clone());

        }

        let receipt = outcome.exit_code.filter(|_| output_error.is_none()).map(|code| Receipt {

            outputs,

            error: output_error.or_else(|| outcome.timed_out.then(|| "stopped at the time limit".to_string())),

- `publish` and `secret` are recognised and refused with "not in M0", so

  a pipeline written for the full spec fails clearly.

- `output { artifact "oxy" { path "/src/target/release/oxy" } }` declares

  up to 16 **artifacts** (spec §5.2). The path is clean, absolute, under

  `/src`, and not on a tmpfs mount, which is gone when the job ends. After

  a job passes, the runner reads each artifact from the job's disks with

  `debugfs`: the workspace, or the `ephemeral://disk` mount holding it. An

  artifact is at most 512 MiB. The runner uploads each one with

  `POST /api/v1/flow/jobs/{id}/artifacts/{name}` (raw bytes). The hub keeps

  it as a blob of the repository, and the receipt's `outputs` name those

  blobs in artifact order. A declared artifact that cannot be kept fails

  the job, and the job gets no receipt.

into an image); mounts other than ephemeral scratch; artifacts as inputs of later jobs; receipt reuse; provenance

attestations; policy gates; secrets; proposal