Artifacts between jobs: input { artifact "x" from "job"; path "/src/x" } 46fe2c815cc3

    /// The job's inputs (oxygenflow#58), in the job's order, each with the

    /// blob the producing job's receipt names: fetched from

    /// `/api/v1/flow/jobs/{job_id}/inputs/{index}` and checked against it.

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

    pub inputs: Vec<ArtifactInput>,

}

/// An artifact a job takes (oxygenflow#58), as the hub hands it over.

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

pub struct ArtifactInput {

    pub artifact: String,

    /// The producing job.

    pub from: String,

    /// Where it goes, under `/src`.

    pub path: String,

    /// The Oxygen blob reference its receipt names, hex.

    pub blob: String,

    // Inputs as [artifact, from, path] sorted by path; absent when none.

    if !j.inputs.is_empty() {

        fields.push((

            10,

            Value::Array(j.inputs.iter().map(|i| Value::Array(vec![text(&i.artifact), text(&i.from), text(&i.path)])).collect()),

        ));

    }

    arch_ok, egress_allows, egress_pattern_ok, Identity, Input, Job, Mount, MountSource, Network, Output, OutputKind, Pipeline, Problem,

    Problems, Runtime, Sandbox, Secret, EGRESS_LABEL, PIPELINE_PATH,

    /// Artifacts of jobs this one depends on, put in its workspace before it

    /// starts (oxygenflow#58), sorted by path. The runner checks each against

    /// the producing job's receipt; this job's receipt lists them as inputs.

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

    pub inputs: Vec<Input>,

}

/// An artifact a job takes from a job it depends on (oxygenflow#58):

/// `input { artifact "tfplan" from "plan"; path "/src/tfplan" }`.

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

pub struct Input {

    /// The producing job's artifact.

    pub artifact: String,

    /// The producing job: one this job depends on.

    pub from: String,

    /// Where it is put: a clean absolute path under `/src`.

    pub path: String,

/// The most inputs a job takes.

const MAX_INPUTS: usize = 8;

                }

            }

            // Inputs come from a job this one depends on, which declares

            // that artifact (oxygenflow#58).

            for i in &job.inputs {

                let Some((producer, _)) = jobs.get(&i.from) else {

                    ctx.err(*span, format!("job {:?} takes an input from {:?}, which is not declared", job.name, i.from));

                    continue;

                };

                if !job.depends_on.contains(&i.from) {

                    ctx.err(*span, format!("job {:?} takes {:?} from {:?}: add depends-on \"{}\"", job.name, i.artifact, i.from, i.from));

                match producer.outputs.iter().find(|o| o.name == i.artifact) {

                    Some(o) if o.kind == OutputKind::Artifact => {}

                    Some(_) => ctx.err(*span, format!("{:?} of job {:?} is a report; inputs are artifacts", i.artifact, i.from)),

                    None => ctx.err(*span, format!("job {:?} declares no artifact {:?}", i.from, i.artifact)),

                }

        }

    }

}

/// `input { artifact "NAME" from "JOB"; path "/src/FILE" }`.

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

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

        ctx.err(node.span(), "`input` takes a block: input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\" }");

    }

    let (mut artifact, mut from, mut path) = (None, None, None);

    for c in block(ctx, node) {

        ctx.no_children(c);

        match c.name().value() {

            "artifact" => match ctx.strings(c).as_slice() {

                [name, word, producer] if word == "from" => {

                    if ctx.name_ok(c.span(), "artifact", name) && ctx.name_ok(c.span(), "job", producer) {

                        artifact = Some(name.clone());

                        from = Some(producer.clone());

                    }

                }

                _ => ctx.err(c.span(), "`artifact` names the artifact and its job: artifact \"tfplan\" from \"plan\""),

            },

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

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

                Some(p) => {

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

                }

                None => {}

            },

            other => ctx.err(c.span(), format!("unknown input setting `{other}`; an input has `artifact` and `path`")),

        }

    }

    match (artifact, from, path) {

        (Some(artifact), Some(from), Some(path)) => {

            if job.inputs.iter().any(|i| i.path == path) {

                ctx.err(node.span(), format!("two inputs are put at {path:?}"));

            } else if job.inputs.len() == MAX_INPUTS {

                ctx.err(node.span(), format!("a job takes at most {MAX_INPUTS} inputs"));

            } else {

                job.inputs.push(Input { artifact, from, path });

            }

        }

        (a, _, p) => {

            if a.is_some() && p.is_none() {

                ctx.err(node.span(), "an input needs a `path`");

            } else if a.is_none() && p.is_some() {

                ctx.err(node.span(), "an input needs `artifact \"NAME\" from \"JOB\"`");

            }

        inputs: vec![],

            "input" => parse_input(ctx, c, &mut job),

    job.inputs.sort_by(|a, b| a.path.cmp(&b.path));

        inputs: vec![],

}

#[test]

fn a_job_takes_an_artifact_of_a_job_it_depends_on() {

    let jobs = |apply: &str| {

        doc(&format!(

            "job \"plan\" {{\n    sandbox \"alpine\"\n    output {{ artifact \"tfplan\" {{ path \"/src/out/tfplan\"; }}; report \"r\" {{ path \"/src/out/r\"; }}; }}\n    exec \"plan\"\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n{apply}    exec \"apply\"\n}}\n"

        ))

    };

    let ok = jobs("    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\"; }\n");

    let p = Pipeline::parse(&ok).unwrap();

    assert_eq!(p.jobs["apply"].inputs, [crate::Input { artifact: "tfplan".into(), from: "plan".into(), path: "/src/tfplan".into() }]);

    let without = Pipeline::parse(&jobs("    depends-on \"plan\"\n")).unwrap();

    assert_ne!(p.hash(), without.hash(), "inputs are part of the job");

    let has = |apply: &str, want: &str| {

        let ps = problems(&jobs(apply));

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

    };

    has("    input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\"; }\n", "add depends-on \"plan\"");

    has("    depends-on \"plan\"\n    input { artifact \"nope\" from \"plan\"; path \"/src/x\"; }\n", "declares no artifact \"nope\"");

    has("    depends-on \"plan\"\n    input { artifact \"r\" from \"plan\"; path \"/src/x\"; }\n", "is a report");

    has("    input { artifact \"tfplan\" from \"ghost\"; path \"/src/x\"; }\n", "which is not declared");

    has("    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; path \"/etc/x\"; }\n", "absolute path under /src");

    has("    depends-on \"plan\"\n    input { artifact \"tfplan\" \"plan\"; path \"/src/x\"; }\n", "artifact \"tfplan\" from \"plan\"");

    has("    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; }\n", "needs a `path`");

    has(

        "    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; path \"/src/x\"; }\n    input { artifact \"tfplan\" from \"plan\"; path \"/src/x\"; }\n",

        "two inputs are put at",

    );

            inputs: vec![],

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

    /// Input `index` of job `job` (oxygenflow#58): the bytes of the

    /// producing job's artifact, which the caller checks against the blob the

    /// claim names.

    pub fn input(&self, job: i64, index: usize) -> Result<Vec<u8>> {

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

        let r = retrying(&url, &PATIENT_DELAYS, false, || {

            self.agent

                .get(&url)

                .timeout(Duration::from_secs(600))

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

                .call()

                .map_err(Box::new)

        })?;

        let mut bytes = Vec::new();

        let mut reader = r.into_reader();

        std::io::Read::read_to_end(&mut std::io::Read::take(&mut reader, MAX_INPUT_BYTES as u64 + 1), &mut bytes)?;

        if bytes.len() > MAX_INPUT_BYTES {

            bail!("larger than {} MiB", MAX_INPUT_BYTES >> 20);

        }

        Ok(bytes)

    }

/// Check an input's bytes against the blob the producing job's receipt

/// names, then write them at its path under `tree` (the workspace's `/src`).

fn place_input(tree: &Path, i: &flow_core::api::ArtifactInput, bytes: &[u8]) -> Result<[u8; 32]> {

    let Some(rel) = i.path.strip_prefix("/src/").filter(|r| r.split('/').all(|p| !p.is_empty() && p != "." && p != ".." && p != ".oxy"))

    else {

        bail!("path {:?} is not a clean path under /src", i.path);

    };

    let blob = Repo::blob_ref(bytes);

    if blob.to_hex() != i.blob {

        bail!("it is not what {}'s receipt names (blob {}, not {})", i.from, &blob.to_hex()[..12], &i.blob[..i.blob.len().min(12)]);

    }

    let dest = tree.join(rel);

    if let Some(dir) = dest.parent() {

        std::fs::create_dir_all(dir)?;

    }

    std::fs::write(&dest, bytes)?;

    Ok(blob.0)

}

/// The largest input a job takes: the hub's artifact limit.

pub const MAX_INPUT_BYTES: usize = 512 << 20;

        // Inputs (oxygenflow#58): each producing job's artifact, checked

        // against the blob its receipt names, then put at its path. Not part

        // of the scope: they are the run's, not the repository's.

        let mut input_blobs = Vec::new();

        for (index, i) in claim.inputs.iter().enumerate() {

            let what = format!("input {} from {}", i.artifact, i.from);

            let bytes = hub.input(claim.job_id, index).with_context(|| format!("{what}: fetching it"))?;

            let blob = place_input(&scratch.join("tree"), i, &bytes).with_context(|| what.clone())?;

            note(&mut sink, format!("{what}: {} bytes at {}, blob {} as its receipt names", bytes.len(), i.path, &i.blob[..12]));

            input_blobs.push(blob);

        }

            inputs: input_blobs.clone(),

    #[test]

    fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {

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

        let bytes = b"the reviewed plan";

        let blob = Repo::blob_ref(bytes).to_hex();

        let input = |path: &str, blob: &str| flow_core::api::ArtifactInput {

            artifact: "tfplan".into(),

            from: "plan".into(),

            path: path.into(),

            blob: blob.into(),

        };

        let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();

        assert_eq!(hex::encode(got), blob);

        assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);

        let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();

        assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");

        assert!(!dir.path().join("other").exists(), "nothing written when it does not match");

        for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {

            assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");

        }

    }

### 5.3.1 Artifacts between jobs (oxygenflow#58)

A job takes an artifact of a job it depends on:

```kdl

job "apply" {

    depends-on "plan"

    input {

        artifact "tfplan" from "plan"

        path "/src/tfplan"

    }

}

```

The producer must be named in `depends-on` and declare that name as an

artifact (not a report); paths are clean absolute paths under `/src`, at

most 8 inputs, and they are part of the job's hash (`[artifact, from,

path]`, field 10). Since the producer has passed when the job is handed

out, the hub puts each input in the claim with the blob the producer's

receipt names (a passing job's receipt names exactly the artifacts it

uploaded). Before the job starts, the runner fetches each from

`GET /api/v1/flow/jobs/{job}/inputs/{index}`, checks its Oxygen blob

reference against that blob, refuses the job if they differ, and writes it

at its path in the workspace (not part of the scope hash: it belongs to the

run, not the repository). The job's receipt lists the blobs as its

`inputs`, so an apply job's receipt proves which plan it applied.