Run pipeline and Re-run: runs that move no bookmark 1edac4b68c6d

-- What started a run: 'push', 'manual' (Run pipeline) or 'rerun:N' (a re-run of run N).

ALTER TABLE flow_runs ADD COLUMN trigger TEXT NOT NULL DEFAULT 'push';

        .route("/repos/{owner}/{name}/flow/runs", post(start_flow_run))

        .route("/repos/{owner}/{name}/flow/runs/{number}/rerun", post(rerun_flow_run))

}

#[derive(Deserialize)]

struct StartRun {

    /// Default: the repository's default bookmark.

    #[serde(default)]

    bookmark: Option<String>,

}

fn started(app: &Shared, row: &db::repos::RepoRow, number: i64) -> (StatusCode, Json<serde_json::Value>) {

    let url = format!("{}/{}/{}/flow/{number}", app.config.base_url(), row.owner, row.name);

    (StatusCode::CREATED, Json(serde_json::json!({ "number": number, "url": url })))

}

/// `POST /repos/{o}/{n}/flow/runs` (repo:write): run a bookmark's current

/// state without moving it (hub spec §13.2).

async fn start_flow_run(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path((owner, name)): Path<(String, String)>,

    Json(body): Json<StartRun>,

) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {

    let c = caller(&app, &headers).await?;

    let me = c.require("repo:write")?.clone();

    let (_, row, _) = reach(&app, &c, &owner, &name, Access::Write).await?;

    let bookmark = body.bookmark.unwrap_or_else(|| row.default_bookmark.clone());

    let n = crate::flow::run_bookmark(&app, &row, &bookmark, &me.username).await.map_err(|m| ApiError(StatusCode::CONFLICT, m))?;

    db::accounts::audit(&app.db, Some(&me), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {bookmark}")).await.ok();

    Ok(started(&app, &row, n))

}

/// `POST /repos/{o}/{n}/flow/runs/{number}/rerun` (repo:write).

async fn rerun_flow_run(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path((owner, name, number)): Path<(String, String, i64)>,

) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {

    let c = caller(&app, &headers).await?;

    let me = c.require("repo:write")?.clone();

    let (_, row, _) = reach(&app, &c, &owner, &name, Access::Write).await?;

    let run = db::flow::run(&app.db, row.id, number).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;

    let n = crate::flow::rerun(&app, &row, &run, &me.username).await.map_err(|m| ApiError(StatusCode::CONFLICT, m))?;

    db::accounts::audit(&app.db, Some(&me), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}")).await.ok();

    Ok(started(&app, &row, n))

    /// `push`, `manual` or `rerun:N` (migration 0020).

    #[sqlx(default)]

    pub trigger: String,

    /// How the run page says what started it.

    pub fn started_by(&self) -> String {

        match self.trigger.as_str() {

            "manual" => format!("started by {}", self.actor),

            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", &t[6..], self.actor),

            _ => format!("pushed by {}", self.actor),

        }

    }

    /// `push`, `manual` or `rerun:N`.

    pub trigger: &'a str,

        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \

         VALUES (?, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",

    .bind(r.trigger)

    if let Err(e) = start_run(app, row, bookmark, state, heads, actor, "push").await {

        tracing::warn!("flow: starting a run of {}/{}: {e}", row.owner, row.name);

    }

}

/// "Run pipeline" (hub spec §13.2): run a bookmark's current state without

/// moving it. Returns the new run's number.

pub async fn run_bookmark(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, actor: &str) -> Result<i64, String> {

    if !enabled(app, row) {

        return Err("OxygenFlow is not enabled for this repository".into());

    }

    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;

    let b = bookmark.to_string();

    let found = app

        .repos

        .read(&repo, move |r, _| {

            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };

            Ok(Some((st, r.store().get_state(&st)?.heads)))

        })

        .await

        .map_err(|e| e.to_string())?;

    let Some((state, heads)) = found else { return Err(format!("no bookmark {bookmark}")) };

    if heads.is_empty() {

        return Err(format!("{bookmark} has no changes yet"));

    }

    start_run(app, row, bookmark, &state.to_hex(), &heads, actor, "manual")

        .await?

        .ok_or_else(|| format!("{bookmark} has no {PIPELINE_PATH}"))

}

/// Re-run a finished run: the same state, bookmark and pipeline, as a new

/// run whose checks replace the old ones. Returns the new run's number.

pub async fn rerun(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<i64, String> {

    if !enabled(app, row) {

        return Err("OxygenFlow is not enabled for this repository".into());

    }

    if run.live() {

        return Err(format!("run #{} is still {}", run.number, run.status));

    }

    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();

    if heads.is_empty() {

        return Err("the run names no changes".into());

    }

    start_run(app, row, &run.bookmark, &run.state, &heads, actor, &format!("rerun:{}", run.number))

        .await?

        .ok_or_else(|| format!("that state has no {PIPELINE_PATH}"))

}

/// Start a run of the pipeline in the state `heads` make (whose state

/// hash is `state`), recorded on `bookmark`: for a push, "Run pipeline"

/// (`trigger` = `manual`) and re-runs (`rerun:N`). `actor` is the account

/// whose access the run's subrepo reads are checked against. Returns the

/// run's number, or `None` when the state has no pipeline. The caller

/// checks that Flow is enabled.

pub async fn start_run(

    app: &Shared,

    row: &db::repos::RepoRow,

    bookmark: &str,

    state: &str,

    heads: &[ChangeHash],

    actor: &str,

    trigger: &str,

) -> Result<Option<i64>, String> {

    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;

        .await

        .map_err(|e| format!("reading the pipeline: {e}"))?;

    let Some(text) = text else { return Ok(None) };

    let base = db::flow::NewRun { repo_id: row.id, bookmark, state, heads: &heads_s, pipeline_hash: "", actor, trigger, error: None };

    let id = match Pipeline::parse(&text) {

            db::flow::create_run(&app.db, db::flow::NewRun { error: Some(&msg), ..base }, &[]).await.map_err(|e| e.to_string())?

            let id =

                db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await.map_err(|e| e.to_string())?;

            if let Ok(Some(run)) = fdb::run_by_id(&app.db, id).await {

    let run = fdb::run_by_id(&app.db, id).await.map_err(|e| e.to_string())?.ok_or("the run vanished")?;

    Ok(Some(run.number))

use super::settings::{csrf, require};

use super::{filters, lookup, lookup_role, Error, Page, RepoShell, Shell};

use crate::auth::{Access, Role, Viewer};

use axum::extract::{Form, Path, Query, State};

use axum::response::{IntoResponse, Redirect, Response};

    /// Writers may start runs: the bookmarks to offer, default first.

    can_run: bool,

    bookmarks: Vec<String>,

    error: Option<String>,

    /// Why a "Run pipeline" or "Re-run" did not start.

    error: Option<String>,

    let can_run = enabled && viewer.account().is_some() && role >= Role::Write;

    let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] };

    Ok(Page(RunsPage {

        shell,

        repo: RepoShell::new(&app, &row, "flow").with_role(role),

        runs,

        enabled,

        older,

        can_run,

        bookmarks,

        error: q.error,

    }))

}

/// The repository's bookmarks, its default first.

async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec<String> {

    let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] };

    let mut names: Vec<String> = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default();

    names.sort_by_key(|n| (n != &row.default_bookmark, n.clone()));

    names

}

#[derive(Deserialize)]

pub struct StartForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    bookmark: String,

}

/// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark.

pub async fn start(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Form(f): Form<StartForm>,

) -> Result<Response, Error> {

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

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    Ok(match crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username).await {

        Ok(n) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {}", f.bookmark.trim()))

                .await

                .ok();

            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()

        }

        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),

    })

}

#[derive(Deserialize)]

pub struct RerunForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

}

/// `POST /{owner}/{name}/flow/{number}/rerun`.

pub async fn rerun(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, number)): Path<(String, String, i64)>,

    Form(f): Form<RerunForm>,

) -> Result<Response, Error> {

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

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

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

        .await

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

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

    Ok(match crate::flow::rerun(&app, &row, &run, &me.account.username).await {

        Ok(n) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}"))

                .await

                .ok();

            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()

        }

        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),

    })

}

/// Percent-encode a message for a query string.

fn urlencoding(s: &str) -> String {

    s.bytes()

        .map(|b| match b {

            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(),

            _ => format!("%{b:02X}"),

        })

        .collect()

    /// Whether the viewer may re-run it.

    can_rerun: bool,

    let can_rerun = !run.live() && viewer.account().is_some() && role >= Role::Write && crate::flow::enabled(&app, &row);

    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun }))

        .route("/{owner}/{name}/flow/start", post(flow::start))

        .route("/{owner}/{name}/flow/{id}/rerun", post(flow::rerun))

/// "Run pipeline" and "Re-run" (hub spec §13.2): runs that move no

/// bookmark, through the API and the pages.

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

async fn runs_start_without_a_push_and_finished_runs_re_run() {

    let h = hub().await;

    // The site token acts as the first site administrator.

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    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", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

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

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

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

    assert_eq!((first.len(), first[0].trigger.as_str()), (1, "push"));

    // Run pipeline on the default bookmark: a new run of the same state.

    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;

    assert_eq!(st, StatusCode::CREATED, "{body}");

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

    assert_eq!(v["number"], 2);

    assert!(v["url"].as_str().unwrap().ends_with("/andy/demo/flow/2"));

    let second = db::flow::run(&h.app.db, first[0].repo_id, 2).await.unwrap().unwrap();

    assert_eq!((second.trigger.as_str(), second.actor.as_str(), second.state.as_str()), ("manual", "andy", first[0].state.as_str()));

    assert_eq!(db::flow::jobs(&h.app.db, second.id).await.unwrap().len(), 3);

    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/2", None, None).await;

    assert!(page.contains("started by andy"), "the run page says what started it");

    // A live run is not re-run; unknown bookmarks and runs are refused.

    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;

    assert!(st == StatusCode::CONFLICT && body.contains("still queued"), "{st} {body}");

    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(r#"{"bookmark":"nope"}"#.into())).await;

    assert_eq!(st, StatusCode::CONFLICT);

    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/99/rerun", Some("site"), None).await;

    assert_eq!(st, StatusCode::NOT_FOUND);

    // Anonymous callers cannot start runs.

    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", None, Some("{}".into())).await;

    assert_ne!(st, StatusCode::CREATED);

    // A finished run re-runs as a new run of the same state.

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

    for r in [&first[0], &second] {

        for j in db::flow::jobs(&h.app.db, r.id).await.unwrap() {

            sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE id = ?").bind(j.id).execute(&h.app.db).await.unwrap();

        }

        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = 1 WHERE id = ?").bind(r.id).execute(&h.app.db).await.unwrap();

    }

    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;

    assert_eq!(st, StatusCode::CREATED, "{body}");

    let third = db::flow::run(&h.app.db, row.id, 3).await.unwrap().unwrap();

    assert_eq!((third.trigger.as_str(), third.status.as_str(), third.state.as_str()), ("rerun:1", "queued", first[0].state.as_str()));

    let checks = db::hooks::checks_for_state(&h.app.db, row.id, &third.state).await.unwrap();

    assert!(

        checks.iter().all(|c| c.status == "queued" && c.url.ends_with("/flow/3")),

        "the re-run's checks replace the old ones: {checks:?}"

    );

    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/3", None, None).await;

    assert!(page.contains("re-run of #1 by andy"));

    assert!(!page.contains("RE-RUN"), "anonymous viewers get no button");

    // With Flow off, nothing starts.

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

    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;

    assert!(st == StatusCode::CONFLICT && body.contains("not enabled"), "{body}");

}

JSON under `/api/v1`, bearer tokens, `application/json`, RFC 7807 problem responses, cursor pagination (`?after=`), ETags on cacheable resources, and rate limits per token. Resources mirror the pages: `/repos/{owner}/{name}`, `/bookmarks`, `/changes/{hash}`, `/tree/{ref}/{path}`, `/file/{ref}/{path}`, `/blame`, `/history`, `/compare`, `/conflicts`, `/proposals`, `/proposals/{n}/comments|reviews|checks|files|land`, `/proposals/{n}/content/{base|head}/{path}` (a changed file's bytes in the target's state or the projected state, for editor diff views), `/issues` (creating an issue can set its labels, assignees and milestone; `PATCH /issues/{n}` edits the title and body for the author or triage, and replaces labels, assignees or the milestone for triage, recording each on the timeline; unknown names are refused before anything changes), `/labels` (`POST` creates or changes one, `DELETE /labels/{name}` removes it; triage), `/milestones`, `/releases`, `/tags`, `/trust`, `/pushes`, `/hooks`, `/checks`, `/user`, `/notifications` (the caller's inbox and unread count; `/notifications/{id}/read` and `/notifications/read` mark read; scope `user:read`), `/users/{name}`, `/orgs/{name}`, `/checks/required` (`GET`, and `PUT` of the whole list by a repository admin: the checks a proposal needs, passed, to land), `/projects/{owner}` (list, create) and `/projects/{owner}/{n}` (the board as the caller may see it, `PATCH` for title, description, privacy and closing, `/cards` to add a reference or note to a column named by id or name, `/cards/{id}` to move or remove one; `repo:write` and the page's rights to change), `/flow/runs` (`POST {"bookmark"}`, repo:write: run a bookmark's current state, the default bookmark when none is named) and `/flow/runs/{n}/rerun` (repo:write: a finished run again), both answering `201 {"number", "url"}` or `409` with why nothing started, `/search`. Shapes reuse `oxy --json` shapes wherever one exists (change, status, blame, diff) so tools written against the CLI work against the hub. `crates/hub-api` holds the types with `serde` derives and is what the runner and the VS Code extension will link.

- **Triggers.** A push that moves a bookmark of an allowed repository starts a **run** on the bookmark's new state, numbered within its repository from 1 (the run's page is `/{owner}/{name}/flow/{number}`): the hub reads `.flow/pipeline.kdl` from that state and validates it with `flow-core` (the same code `flow check` runs). An invalid pipeline records a run in `error` with the problems and their lines; a state without a pipeline records nothing. Tags and proposals start no runs yet, so only code from people with write access ever runs. Writers also start runs without moving a bookmark: **Run pipeline** on the Flow tab runs a bookmark's current state, and **Re-run** on a finished run's page runs its state again. The new run's checks replace the old ones for that state, and the run page says what started it (pushed by, started by, re-run of #N by).

  <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> · {{ run.started_by() }} <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>

  {% if can_rerun %}<form class="inline" method="post" action="{{ repo.url }}/flow/{{ run.number }}/rerun"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="small">RE-RUN</button></form>{% endif %}

<div class="head"><h1>Flow <small>pipeline runs</small></h1>

{% if can_run %}<form class="inline" method="post" action="{{ repo.url }}/flow/start"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><label for="run-bookmark" class="small mute">Run the pipeline of</label> <select id="run-bookmark" name="bookmark">{% for b in bookmarks %}<option value="{{ b }}">{{ b }}</option>{% endfor %}</select> <button type="submit" class="small">RUN PIPELINE</button></form>{% endif %}

</div>

{% if let Some(e) = error %}<p class="error">Could not start a run: {{ e }}</p>{% endif %}