Number Flow runs per repository a0d4bbfbedd1

pub struct Run {

    pub id: i64,

    pub repo_id: i64,

    pub bookmark: String,

    pub state: String,

    pub heads: String,

    let now = now_ms();

    let status = if r.error.is_some() { "error" } else { "queued" };

    let id: i64 = sqlx::query_scalar(

    )

    .bind(r.repo_id)

    .bind(r.bookmark)

    .bind(r.state)

    .bind(r.heads)

        .await

}

}

pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {

                .collect();

            let hash = hex::encode(p.hash());

            let id = db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await;

                for j in &jobs {

                }

            }

            id

    }

}

    let name = format!("flow/{job}");

    let c = db::hooks::upsert_check(

        &app.db,

    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 bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));

    let claim = Claim {

        job_id: job.id,

                    "skipped" => "skipped: a dependency did not pass".into(),

                    _ => format!("could not run: {}", j.error),

                };

            }

        }

    }

    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;

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

    let to = match run {

        None => format!("/{owner}/{name}/flow"),

    };

    Ok(axum::response::Redirect::to(&to).into_response())

    }

    let meta = serde_json::json!({

        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),

        "status": status,

        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),

    });

    let title = match run {

        None => format!("OxygenFlow: no run for state {short}"),

    };

    format!(

        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),

        [("flow/build", "passed"), ("flow/lint", "passed"), ("flow/test", "failed")]

    );

    // Pages: the list, the run with its graph and logs, and the live view.

    let (st, list) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;

    assert_eq!(st, StatusCode::OK);

    assert_eq!(st, StatusCode::OK);

    assert!(page.contains("<svg class=\"flow-graph\""), "graph");

    assert_eq!(page.matches("data-job-node=").count(), 3);

    assert_eq!(page.matches("<path class=\"edge").count(), 1, "one dependency edge");

    assert!(page.contains("cc -o demo demo.c"), "log shown");

    assert!(!page.contains("data-flow-live"), "a finished run does not poll");

    assert_eq!(st, StatusCode::OK);

    let live: serde_json::Value = serde_json::from_str(&live).unwrap();

    assert_eq!(live["status"], "failed");

    let third = third.expect("a third run");

    assert_eq!(third.status, "error");

    assert!(third.error.contains("line 18") && third.error.contains("\"missing\""), "{}", third.error);

    assert!(page.contains("The pipeline is not valid"));

    // State stamps (hub spec §13.2) and images in READMEs.

    let h = hub().await;

    create(&h.app, "other", "plain").await;

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

    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();

    db::repos::set_flow(&h.app.db, row.id, "on").await.unwrap();

    let quiet = db::repos::get(&h.app.db, "andy", "quiet").await.unwrap().unwrap();

        c.commit("Add a pipeline", &[]).unwrap();

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

    }

    assert!(runs_of(&h.app, "andy", "quiet").await.is_empty(), "turned off despite andy/*");

}

- **Configuration.** `[flow] runner_tokens` (bearer tokens for runners, compared in constant time), `repos` (patterns such as `owner/*` or `owner/name`: the repositories whose pushes start runs by default; empty means none) and `job_timeout_minutes` (60).

- **Per repository.** A site administrator turns OxygenFlow on or off for one repository under Settings → OxygenFlow, which overrides `repos` in either direction; "site default" follows it again. Runs use the site's runners, so repository administrators see the setting but cannot change it. Changes are audited (`repo.flow`) and apply from the next push, with no restart.

- **Runs and jobs.** A run holds the repository, bookmark, state hash and heads, pipeline hash and pusher; one job per pipeline job with its compiled form, sandbox and dependencies. Job states: `queued`, `running`, `passed`, `failed`, `skipped` (a dependency did not pass), `error` (the runner could not run it). A run is `running` while any job is, then `passed` when every job passed, else `failed`.

- **Runner API** (runner token): `POST /api/v1/flow/claim` hands out the oldest queued job whose dependencies passed and whose tier the runner offers, marking it `running`, with a **job token**; `204` when there is none. `POST /api/v1/flow/jobs/{id}/log` appends log records (text cut at 16 KiB, at most 50 000 per job); `POST /api/v1/flow/jobs/{id}/finish` records the exit code and receipt (the hub recomputes the receipt hash) and settles dependents. Only the runner holding the job may report on it.

- **Job tokens.** Random, stored as a hash, valid only for reading the job's repository over the wire protocol while the job is running and for at most two hours. They cannot push, call the API or read other repositories.

{% extends "base.html" %}

{% import "repo_tabs.html" as m %}

{% block tabs %}{% call m::tabs(repo, "flow") %}{% endblock %}

{% block mainclass %} class="wide"{% endblock %}

{% block content %}

<div class="head">

  <p class="small mute"><span class="lbl lbl-bm">{{ run.bookmark }}</span> at <a href="{{ repo.url }}/tree/{{ run.heads }}" title="state {{ run.state }}"><code>{{ run.state|short }}</code></a> · pushed by {{ run.actor }} <time title="{{ run.created_ms|when_i }}">{{ run.created_ms|ago_i }}</time> · {{ duration }}{% if !run.pipeline_hash.is_empty() %} · pipeline <code title="{{ run.pipeline_hash }}">{{ run.pipeline_hash|short }}</code>{% endif %} · <a href="{{ repo.url }}/file/{{ run.heads }}/.flow/pipeline.kdl">.flow/pipeline.kdl</a> · <a href="{{ repo.url }}/flow">all runs</a></p>

</div>

{% if run.status == "error" %}

<section class="panel"><div class="strip bad">The pipeline is not valid, so nothing ran.</div><pre class="snippet">{{ run.error }}</pre><p class="small mute">Run <code>flow check</code> locally to see the same problems.</p></section>

{% else %}

<section class="panel flow-graph-box">

    {% for e in graph.edges %}<path class="edge st-{{ e.status }}" d="{{ e.d }}"/>{% endfor %}

    {% for n in graph.nodes %}

    <a href="#job-{{ n.j.id }}" data-job-node="{{ n.j.id }}" class="node st-{{ n.j.status }}">

  {% for row in runs %}

  <tr class="st-{{ row.r.status }}">

    <td class="st"><span class="flow-glyph st-{{ row.r.status }}" title="{{ row.r.status }}">{{ crate::web::flow::glyph(row.r.status) }}</span></td>

    <td class="flow-dots">{% for (n, s) in row.jobs %}<span class="flow-glyph st-{{ s }}" title="{{ n }}: {{ s }}">{{ crate::web::flow::glyph(s) }}</span>{% endfor %}{% if row.r.status == "error" %}<span class="small bad">pipeline invalid</span>{% endif %}</td>

    <td><span class="lbl lbl-bm">{{ row.r.bookmark }}</span></td>

    <td><a href="{{ repo.url }}/tree/{{ row.r.heads }}" title="state {{ row.r.state }}"><code>{{ row.r.state|short }}</code></a></td>