Keep and serve Flow job artifacts a3c8288b0141

-- A job's artifacts (OxygenFlow spec §5.2): each kept as a blob of the

-- run's repository, with its size and the BLAKE3 of its bytes, which is

-- how a release asset finds the receipt that built it.

CREATE TABLE flow_artifacts (

    job_id INTEGER NOT NULL REFERENCES flow_jobs(id) ON DELETE CASCADE,

    name TEXT NOT NULL,

    path TEXT NOT NULL,

    blob TEXT NOT NULL,

    size INTEGER NOT NULL,

    content TEXT NOT NULL,

    created_ms INTEGER NOT NULL,

    PRIMARY KEY (job_id, name)

);

CREATE INDEX flow_artifacts_content ON flow_artifacts(content);

        .route(

            "/flow/jobs/{id}/artifacts/{name}",

            post(crate::flow::artifact).layer(axum::extract::DefaultBodyLimit::max(crate::flow::MAX_ARTIFACT_BYTES + (1 << 20))),

        )

/// A job's artifact, kept as a blob of the run's repository.

#[derive(Clone, Debug, FromRow)]

pub struct Artifact {

    pub job_id: i64,

    pub name: String,

    pub path: String,

    pub blob: String,

    pub size: i64,

    /// BLAKE3 of the bytes.

    pub content: String,

    pub created_ms: i64,

}

pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {

    sqlx::query("INSERT OR REPLACE INTO flow_artifacts (job_id, name, path, blob, size, content, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?)")

        .bind(a.job_id)

        .bind(&a.name)

        .bind(&a.path)

        .bind(&a.blob)

        .bind(a.size)

        .bind(&a.content)

        .bind(a.created_ms)

        .execute(pool)

        .await?;

    Ok(())

}

/// Job `job_id`'s artifacts, by name.

pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {

    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = ? ORDER BY name").bind(job_id).fetch_all(pool).await

}

/// Largest artifact the hub keeps; the runner refuses larger ones first.

pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;

/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact

/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's

/// repository. Answers `{"blob"}`.

pub async fn artifact(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path((id, name)): Path<(i64, String)>,

    body: axum::body::Bytes,

) -> Result<Response, ApiError> {

    runner_auth(&app, &headers)?;

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

    let spec: flow_core::Job =

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

    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", 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 content = blake3::hash(&body).to_hex().to_string();

    let size = body.len() as i64;

    let blob = app

        .repos

        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))

        .await

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

    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };

    fdb::add_artifact(&app.db, &a).await?;

    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())

}

            // The receipt names exactly the artifacts uploaded, in name order.

            let kept: Vec<String> = fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect();

            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {

                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));

            }

    pub artifacts: Vec<fdb::Artifact>,

        let artifacts = fdb::artifacts(&app.db, j.id).await.unwrap_or_default();

        logs.push(JobLog { duration: crate::flow::duration(j), j: j.clone(), artifacts, lines, last_seq });

        ],

        bytes,

    )

        .into_response())

}

/// `GET /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}`: an artifact

/// of a run's job, read from the repository.

pub async fn artifact_file(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}"), &viewer);

    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let run = fdb::run(&app.db, row.id, id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };

    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };

    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };

    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?;

    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();

    Ok((

        [

            (axum::http::header::CONTENT_TYPE, "application/octet-stream".to_string()),

            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}\"", file.replace('"', ""))),

        .route("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}", get(flow::artifact_file))

async fn upload(app: &Shared, job: i64, name: &str, bytes: &[u8]) -> (StatusCode, String) {

    let req = Request::post(format!("/api/v1/flow/jobs/{job}/artifacts/{name}"))

        .header(header::AUTHORIZATION, "Bearer runner-secret")

        .header(header::CONTENT_TYPE, "application/octet-stream")

        .body(Body::from(bytes.to_vec()))

        .unwrap();

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

    let status = res.status();

    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

}

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

async fn artifacts_are_kept_named_by_the_receipt_and_served() {

    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 \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"

        ),

    );

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

    c.commit("Build a tool", &[]).unwrap();

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

    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

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

    assert_eq!(a["job"]["outputs"], serde_json::json!([{ "name": "tool", "path": "/src/out/tool" }]), "the runner is told");

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

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

    let (st, text) = upload(&h.app, id, "other", &body).await;

    assert_eq!(st, StatusCode::BAD_REQUEST, "an undeclared artifact: {text}");

    let (st, text) = upload(&h.app, id, "tool", &body).await;

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

    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();

    assert_eq!(blob, oxy_core::repo::Repo::blob_ref(&body).to_hex(), "the same blob the runner computes");

    // The receipt must name exactly the uploaded artifacts.

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

    assert_eq!(

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

        StatusCode::BAD_REQUEST

    );

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

    assert_eq!(finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r })).await, StatusCode::NO_CONTENT);

    // The run page lists it; anyone who can read the repository downloads it.

    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;

    let link = format!("/andy/demo/flow/{}/artifacts/build/tool", run.number);

    assert!(page.contains(&link), "run page lists the artifact");

    let (st, got) = get_bytes(&h.app, &link).await;

    assert_eq!((st, got == body), (StatusCode::OK, true));

    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/flow/{}/artifacts/build/none", run.number)).await.0, StatusCode::NOT_FOUND);

}

- **Artifacts** (OxygenFlow spec §5.2): a job may declare artifacts with `output`.

  - The runner holding the job uploads each one with `POST /api/v1/flow/jobs/{id}/artifacts/{name}` (raw bytes, at most 512 MiB).

  - The hub refuses names the job does not declare. It keeps each artifact as a blob of the run's repository, records its path, size and BLAKE3 in `flow_artifacts`, and answers with the blob's ObjectRef.

  - `finish` refuses a receipt whose `outputs` are not exactly the uploaded blobs, in name order.

  - The run page lists each job's artifacts, and `/{owner}/{name}/flow/{n}/artifacts/{job}/{name}` downloads one.

  - A kept receipt keeps its artifacts through garbage collection (Oxygen spec §14.6).

  <summary><span class="flow-glyph st-{{ l.j.status }}" data-glyph>{{ crate::web::flow::glyph(l.j.status) }}</span> <strong>{{ l.j.name }}</strong> <span class="small mute" data-meta>{{ l.j.status }}{% if let Some(c) = l.j.exit_code %} · exit {{ c }}{% endif %} · {{ l.duration }}{% if !l.j.runner.is_empty() %} · on {{ l.j.runner }}{% endif %}</span>{% if !l.j.receipt_attest.is_empty() %} <span class="small mute">· <a href="{{ repo.url }}/receipts/{{ l.j.receipt_attest }}" title="Receipt {{ l.j.receipt_hash }}, kept in the repository; check it with oxy receipt verify">signed receipt</a> by {{ l.j.receipt_signer }}</span>{% else if !l.j.receipt_note.is_empty() %} <span class="small warn" title="{{ l.j.receipt_note }}">· receipt not kept: {{ l.j.receipt_note }}</span>{% else if !l.j.receipt_hash.is_empty() %} <span class="small mute">· unsigned receipt <code title="{{ l.j.receipt_hash }}">{{ l.j.receipt_hash|short }}</code></span>{% endif %}{% if !l.artifacts.is_empty() %} <span class="small mute">· {% for a in l.artifacts %}<a href="{{ repo.url }}/flow/{{ run.number }}/artifacts/{{ l.j.name }}/{{ a.name }}" title="{{ a.path }}, blob {{ a.blob }}">{{ a.name }}</a> ({{ a.size|bytes }}){% if !loop.last %}, {% endif %}{% endfor %}</span>{% endif %}</summary>