Artifacts between jobs: the claim names each input's blob, runners fetch inputs cd437c4f5f8d

        .route("/flow/jobs/{id}/inputs/{index}", get(crate::flow::input))

        inputs: vec![],

    // Artifacts of jobs it depends on (oxygenflow#58), with the blob each

    // producer's receipt names; the runner checks what it fetches against it.

    match artifact_inputs(&app, &job, &claim.job).await {

        Ok(found) => claim.inputs = found,

        Err(why) => {

            let error = format!("inputs: {why}");

            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })

                .await?;

            return Err(ApiError(StatusCode::CONFLICT, error));

        }

    }

}

/// The artifacts a job takes (oxygenflow#58), each from a job of the same

/// run that passed with a receipt; a passing job's receipt names exactly the

/// artifacts it uploaded (checked at finish), so their blobs are attested.

async fn artifact_inputs(app: &Shared, job: &fdb::Job, spec: &flow_core::Job) -> Result<Vec<flow_core::api::ArtifactInput>, String> {

    if spec.inputs.is_empty() {

        return Ok(vec![]);

    }

    let jobs = fdb::jobs(&app.db, job.run_id).await.map_err(|e| e.to_string())?;

    let mut out = Vec::new();

    for i in &spec.inputs {

        let Some(producer) = jobs.iter().find(|j| j.name == i.from) else {

            return Err(format!("{:?} is not a job of this run", i.from));

        };

        if producer.status != "passed" || producer.receipt_hash.is_empty() {

            return Err(format!("{} has not passed with a receipt, so {} cannot be taken from it", i.from, i.artifact));

        }

        let artifacts = fdb::artifacts(&app.db, producer.id).await.map_err(|e| e.to_string())?;

        let Some(a) = artifacts.iter().find(|a| a.name == i.artifact) else {

            return Err(format!("{} kept no artifact {}", i.from, i.artifact));

        };

        out.push(flow_core::api::ArtifactInput {

            artifact: i.artifact.clone(),

            from: i.from.clone(),

            path: i.path.clone(),

            blob: a.blob.clone(),

        });

    }

    Ok(out)

}

/// `GET /api/v1/flow/jobs/{id}/inputs/{index}`: the bytes of input `index`

/// of a running job (oxygenflow#58), for the runner holding it.

pub async fn input(State(app): State<Shared>, headers: HeaderMap, Path((id, index)): Path<(i64, usize)>) -> Result<Response, ApiError> {

    let runner = runner_auth(&app, &headers).await?;

    let job = running_job(&app, id).await?;

    // A registered runner reads only the inputs of a job it holds (a

    // configured token's runner names itself when it claims).

    if runner.source == "registered" && job.runner != runner.name {

        return Err(ApiError(StatusCode::FORBIDDEN, "this job is held by another runner".into()));

    }

    let spec: flow_core::Job =

        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;

    let found = artifact_inputs(&app, &job, &spec).await.map_err(|why| ApiError(StatusCode::CONFLICT, why))?;

    let Some(i) = found.get(index) else {

        return Err(ApiError(StatusCode::NOT_FOUND, format!("job {} has no input {index}", job.name)));

    };

    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;

    let row =

        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;

    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;

    let blob = oxy_core::hash::ObjectRef::from_hex(&i.blob)

        .ok_or_else(|| ApiError(StatusCode::INTERNAL_SERVER_ERROR, "a kept artifact has a bad blob".into()))?;

    let bytes = app

        .repos

        .read(&repo, move |r, _| Ok(r.read_blob(&blob)?))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("read the artifact: {e}")))?;

    Ok(([(axum::http::header::CONTENT_TYPE, "application/octet-stream")], bytes).into_response())

#[tokio::test(flavor = "multi_thread")]

async fn a_job_takes_the_artifact_its_producer_kept() {

    let h = hub().await;

    create(&h.app, "andy", "demo").await;

    let url = format!("{}/andy/demo", h.base);

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

    let c = client(&url, work.path());

    write(

        work.path(),

        ".flow/pipeline.kdl",

        &format!(

            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"plan\" {{\n    sandbox \"alpine\"\n    exec \"plan\"\n    output {{\n        artifact \"tfplan\" {{\n            path \"/src/out/tfplan\"\n        }}\n    }}\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n    depends-on \"plan\"\n    input {{\n        artifact \"tfplan\" from \"plan\"\n        path \"/src/tfplan\"\n    }}\n    exec \"apply\"\n}}\n"

        ),

    );

    c.snapshot("test").unwrap();

    c.commit("Plan, then apply that plan", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The plan job runs and keeps its plan, named by its receipt.

    let runs = runs_of(&h.app, "andy", "demo").await;

    assert_eq!(runs.len(), 1, "a run");

    let jobs = db::flow::jobs(&h.app.db, runs[0].id).await.unwrap();

    assert!(!jobs.is_empty(), "jobs: {:?}", runs[0]);

    let p = claim(&h.app)

        .await

        .unwrap_or_else(|| panic!("nothing to claim: {:?}", jobs.iter().map(|j| (&j.name, &j.status, &j.error)).collect::<Vec<_>>()));

    assert_eq!(p["job"]["name"], "plan");

    let plan_id = p["job_id"].as_i64().unwrap();

    let plan = b"the reviewed plan".to_vec();

    let (st, text) = upload(&h.app, plan_id, "tfplan", &plan).await;

    assert_eq!(st, StatusCode::OK, "{text}");

    let blob = oxy_core::repo::Repo::blob_ref(&plan).to_hex();

    let mut r = receipt("plan", p["pipeline_hash"].as_str().unwrap(), 0);

    r["outputs"] = serde_json::json!([blob]);

    assert_eq!(

        finish(&h.app, plan_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r })).await,

        StatusCode::NO_CONTENT

    );

    // The apply job is told what to take and the blob to expect, and the

    // runner fetches exactly those bytes.

    let a = claim(&h.app).await.unwrap();

    assert_eq!(a["job"]["name"], "apply");

    let apply_id = a["job_id"].as_i64().unwrap();

    assert_eq!(

        a["inputs"],

        serde_json::json!([{ "artifact": "tfplan", "from": "plan", "path": "/src/tfplan", "blob": blob }]),

        "the claim names the producer's blob"

    );

    let req = Request::get(format!("/api/v1/flow/jobs/{apply_id}/inputs/0")).header(header::AUTHORIZATION, "Bearer runner-secret");

    let res = oxygenhub::router(h.app.clone()).oneshot(req.body(Body::empty()).unwrap()).await.unwrap();

    assert_eq!(res.status(), StatusCode::OK);

    assert_eq!(res.into_body().collect().await.unwrap().to_bytes().to_vec(), plan);

    let (st, _) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{apply_id}/inputs/1"), Some("runner-secret"), None).await;

    assert_eq!(st, StatusCode::NOT_FOUND, "no second input");

    let (st, _) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{apply_id}/inputs/0"), None, None).await;

    assert_eq!(st, StatusCode::UNAUTHORIZED, "runners only");

    let (st, _) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{plan_id}/inputs/0"), Some("runner-secret"), None).await;

    assert_eq!(st, StatusCode::CONFLICT, "only a running job's");

}