Newest run wins: a proposal's new run supersedes its older ones a9f1c0bf403a

-- Newest run wins (oxygenhub#267): a new push or proposal run cancels the

-- older ones of its bookmark or proposal still live. On, those already

-- running finish and only the queued ones are cancelled; off (the

-- default), running ones are cancelled too.

ALTER TABLE repositories ADD COLUMN flow_keep_running BOOLEAN NOT NULL DEFAULT FALSE;

-- Newest run wins (oxygenhub#267): a new push or proposal run cancels the

-- older ones of its bookmark or proposal still live. On, those already

-- running finish and only the queued ones are cancelled; off (the

-- default), running ones are cancelled too.

ALTER TABLE repositories ADD COLUMN flow_keep_running INTEGER NOT NULL DEFAULT 0;

- **Newest run wins (#267):** a new push run cancels the older push runs of its bookmark still going, as before, and now a proposal's new run cancels that proposal's older runs too, so `proposal/N` runs no longer pile up three or four deep. Each cancelled job says which run superseded it. Runs started by hand, by a schedule, by a tag or as a re-run are never cancelled this way. Settings › OxygenFlow › Newest run wins lets maintainers keep superseded runs that are already running, cancelling only the queued ones.

    /// `push`, `proposal`, `tag:NAME`, `manual`, `schedule` or

    /// `rerun:N:EVENT` (migration 0020).

    /// `push`, `proposal`, `tag:NAME`, `manual`, `schedule` or

    /// `rerun:N:EVENT`.

/// The runs of `bookmark` started by `trigger` still going, older than run

/// `number`: those a newer run supersedes (#267). With `running_too`

/// false, only those still queued.

pub async fn live_runs_before(

    pool: &Pool,

    repo_id: i64,

    bookmark: &str,

    trigger: &str,

    number: i64,

    running_too: bool,

) -> sqlx::Result<Vec<Run>> {

        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND number < $3 AND trigger = $4

         AND (status = 'queued' OR ($5 AND status = 'running')) ORDER BY id",

    .bind(trigger)

    .bind(running_too)

    /// Newest run wins (#267): a superseded run already running finishes;

    /// only queued ones are cancelled. Off by default.

    #[sqlx(default)]

    pub flow_keep_running: bool,

    Ok(())

}

/// Whether a new run leaves the older runs it supersedes that are already

/// running to finish (#267), cancelling only the queued ones.

pub async fn set_flow_keep_running(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {

    sqlx::query("UPDATE repositories SET flow_keep_running = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;

            // It supersedes the push runs still going on the bookmark (#267).

            start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).await

}

/// Newest run wins (oxygenhub#267): whether a new run started by `trigger`

/// cancels the older runs of its bookmark still live.

///

/// - A **push** run supersedes the older push runs of its bookmark: they

///   test states the bookmark has moved past. Diffusion's bumps start push

///   runs too.

/// - A **proposal** run (bookmark `proposal/N`, started when the

///   proposal's target moved, `web::proposals::settle_after_push`)

///   supersedes the older proposal runs of the same proposal: they test

///   projected states it no longer has. It leaves the proposal's source

///   bookmark's push runs alone: those test the source head, a different

///   state that other proposals or the bookmark's own checks may want.

/// - **Tag**, **manual** ("Run pipeline" on a bookmark or a change),

///   **schedule** and **rerun** runs were each asked for in their own

///   right (a person or a cron), so they supersede nothing and nothing

///   supersedes them: a rerun of an old push run tests the state it names,

///   on purpose. They are cancelled by hand.

///

/// Only runs of the same trigger on the same bookmark are superseded, so a

/// push never cancels a rerun or a proposal's run, nor the other way round.

fn supersedes(trigger: &str) -> bool {

    matches!(trigger, "push" | "proposal")

/// Cancel what run `number` supersedes ([`supersedes`]): the older runs of

/// `bookmark` started by the same `trigger` still queued, and those already

/// running unless the repository keeps them (`flow_keep_running`, Settings

/// › OxygenFlow). A push run's scope reaches back past them either way

/// ([`fdb::latest_on`]).

async fn supersede(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, trigger: &str, number: i64) {

    let Ok(older) = fdb::live_runs_before(&app.db, row.id, bookmark, trigger, number, !row.flow_keep_running).await else { return };

    if supersedes(trigger) {

        supersede(app, row, bookmark, trigger, run.number).await;

    }

        .route("/{owner}/{name}/settings/flow-supersede", post(repo_settings::flow_supersede))

pub struct SupersedeForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `on` when the box is ticked: superseded runs already running are

    /// cancelled too.

    #[serde(default)]

    running: String,

}

/// `POST /{owner}/{repo}/settings/flow-supersede` (oxygenhub#267): whether

/// a new push or proposal run cancels the older runs it supersedes that are

/// already running, or only those still queued; for maintainers, audited.

pub async fn flow_supersede(

    State(app): State<Shared>,

    viewer: Viewer,

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

    Form(f): Form<SupersedeForm>,

) -> Result<Page<RepoSettingsPage>, Error> {

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

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

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

    let keep = f.running != "on";

    db::repos::set_flow_keep_running(&app.db, row.id, keep).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, viewer.account(), "repo.flow_keep_running", &format!("{owner}/{name}"), if keep { "on" } else { "off" })

        .await

        .ok();

    let row = db::repos::get_by_id(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?.unwrap_or(row);

    let notice = if keep {

        "A new push or proposal run now cancels only the older runs still queued; those already running finish."

    } else {

        "A new push or proposal run now cancels the older runs it supersedes, queued or running."

    };

    settings_page(&app, shell, row, role, Some(notice.into()), None).await

}

#[derive(Deserialize)]

/// oxygenhub#267, newest run wins: a proposal's new run supersedes its

/// older runs still live, as a push run does its bookmark's; a repository

/// that keeps running runs has only the queued ones cancelled; manual

/// runs and reruns supersede nothing and nothing supersedes them.

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

async fn newest_run_wins() {

    use oxy_core::hash::ChangeHash;

    let h = hub().await;

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

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

    let url = format!("{}/andy/newest", 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("Start", &[]).unwrap();

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

    let pushed = runs_of(&h.app, "andy", "newest").await[0].clone();

    assert_eq!((pushed.number, pushed.trigger.as_str()), (1, "push"));

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

    let start = |row: db::repos::RepoRow, bookmark: &'static str, trigger: &'static str| {

        let (app, state, heads) = (h.app.clone(), pushed.state.clone(), heads.clone());

        async move {

            oxygenhub::flow::start_run_scoped(&app, &row, bookmark, &state, &heads, "andy", trigger, None).await.unwrap().expect("a run")

        }

    };

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

    let run = |n: i64| {

        let (app, id) = (h.app.clone(), row.id);

        async move { db::flow::run(&app.db, id, n).await.unwrap().unwrap() }

    };

    let superseded_by = |n: i64, by: i64| {

        let app = h.app.clone();

        let run = run(n);

        async move {

            let r = run.await;

            assert_eq!(r.status, "cancelled", "run #{n} is superseded");

            let jobs = db::flow::jobs(&app.db, r.id).await.unwrap();

            let why = format!("superseded by run #{by}");

            assert!(jobs.iter().all(|j| j.status == "cancelled" && j.error == why), "{jobs:?}");

        }

    };

    // A proposal's new run supersedes its older one, not another

    // proposal's, nor the push run of a bookmark.

    assert_eq!(start(row.clone(), "proposal/7", "proposal").await, 2);

    assert_eq!(start(row.clone(), "proposal/7", "proposal").await, 3);

    superseded_by(2, 3).await;

    assert_eq!(start(row.clone(), "proposal/8", "proposal").await, 4);

    assert!(run(3).await.live(), "another proposal's run is not superseded");

    assert!(run(1).await.live(), "a proposal's run leaves push runs alone");

    // Manual runs and reruns supersede nothing, and a push supersedes

    // only push runs.

    assert_eq!(start(row.clone(), "main", "manual").await, 5);

    assert_eq!(start(row.clone(), "main", "rerun:1:push").await, 6);

    assert!(run(1).await.live(), "a manual run or a rerun supersedes nothing");

    assert_eq!(start(row.clone(), "main", "push").await, 7);

    superseded_by(1, 7).await;

    assert!(run(5).await.live() && run(6).await.live(), "a push leaves manual runs and reruns alone");

    // Kept running: a superseded run already running finishes; queued

    // ones are still cancelled.

    db::repos::set_flow_keep_running(&h.app.db, row.id, true).await.unwrap();

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

    assert!(kept.flow_keep_running);

    let three = run(3).await;

    sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = $1").bind(three.id).execute(&h.app.db).await.unwrap();

    sqlx::query("UPDATE flow_jobs SET status = 'running' WHERE run_id = $1 AND name = 'build'")

        .bind(three.id)

        .execute(&h.app.db)

        .await

        .unwrap();

    assert_eq!(start(kept.clone(), "proposal/7", "proposal").await, 8);

    assert_eq!(start(kept.clone(), "proposal/7", "proposal").await, 9);

    assert_eq!(run(3).await.status, "running", "a running run is kept");

    superseded_by(8, 9).await;

    // The default again: running runs are cancelled too.

    db::repos::set_flow_keep_running(&h.app.db, row.id, false).await.unwrap();

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

    assert_eq!(start(row.clone(), "proposal/7", "proposal").await, 10);

    superseded_by(3, 10).await;

    superseded_by(9, 10).await;

}

- **Newest run wins** (oxygenhub#267). A new run cancels the older runs of the same thing still live, so runs do not pile up behind states that have moved on: a push run cancels the older push runs of its bookmark (Diffusion's bumps start push runs), and a proposal's run (bookmark `proposal/N`, started when its target moves) the older runs of that proposal. Each cancelled job says `superseded by run #N`, and the next push run's scope reaches back past what was cancelled. A proposal's run leaves its source bookmark's push runs alone: they test the source head, another state, whose checks the bookmark and other proposals may want. Tag, manual, scheduled and re-run runs were each asked for in their own right, so they supersede nothing and nothing supersedes them; they are cancelled by hand. A repository's maintainers choose under Settings › OxygenFlow › Newest run wins whether superseded runs already running are cancelled too (the default) or finish, with only the queued ones cancelled (`repositories.flow_keep_running`, audited as `repo.flow_keep_running`).

<form class="card wide" id="flow-supersede" method="post" action="{{ repo.url }}/settings/flow-supersede">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>Newest run wins</h2>

  <p>A new push run cancels the older push runs of its bookmark still going, and a proposal's new run its older ones: they test states that have moved on, and each cancelled job says which run superseded it. Runs started by hand, by a schedule, by a tag or as a re-run are never cancelled this way.</p>

  <label class="check"><input type="checkbox" name="running"{% if !row.flow_keep_running %} checked{% endif %}> Cancel superseded runs already running too (untick to let them finish and cancel only those still queued)</label>

  <button type="submit">SAVE</button>

</form>