Cancel a landed proposal's runs that have not started 2abea30809dd

/// The runs on `bookmark` that no runner has started yet: queued, with

/// the given trigger, created before `before_ms` (#263).

pub async fn unstarted_runs_on(pool: &Pool, repo_id: i64, bookmark: &str, trigger: &str, before_ms: i64) -> sqlx::Result<Vec<Run>> {

    sqlx::query_as(

        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND trigger = $3 AND status = 'queued' AND created_ms <= $4 ORDER BY id",

    )

    .bind(repo_id)

    .bind(bookmark)

    .bind(trigger)

    .bind(before_ms)

    .fetch_all(pool)

    .await

}

/// A proposal landed (#263): its runs that no runner has started can no

/// longer change anything, and they hold the queue ahead of the target's

/// own run, so they are cancelled. Its source bookmark's push runs (when

/// that bookmark is in this repository and is not the target) and its

/// `proposal/N` runs; running ones are left to finish.

pub async fn moot_after_landing(app: &Shared, row: &db::repos::RepoRow, p: &db::proposals::ProposalRow) {

    let now = db::now_ms();

    let mut moot = Vec::new();

    let same_repo = p.source_repo_id.is_none_or(|id| id == row.id);

    if p.source_kind == "bookmark" && same_repo && !p.source_bookmark.is_empty() && p.source_bookmark != p.target {

        moot.extend(fdb::unstarted_runs_on(&app.db, row.id, &p.source_bookmark, "push", now).await.unwrap_or_default());

    }

    moot.extend(fdb::unstarted_runs_on(&app.db, row.id, &format!("proposal/{}", p.number), "proposal", now).await.unwrap_or_default());

    for run in moot {

        if let Err(e) = cancel_with(app, &run, &format!("moot: proposal #{} landed", p.number)).await {

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

        }

    }

}

                crate::flow::moot_after_landing(app, row, &l.row).await;

        crate::flow::moot_after_landing(app, &row, &loaded.row).await;

    crate::flow::moot_after_landing(app, row, &p).await;

/// A landed proposal's runs that no runner has started are cancelled

/// (#263): they hold the queue ahead of the target's own run. A running

/// one finishes; another bookmark's run is untouched.

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

async fn landing_cancels_the_proposals_unstarted_runs() {

    let h = hub().await;

    let app = &h.app;

    let andy = register(app, "andy").await;

    let repo = RepoName::parse("andy/demo").unwrap();

    app.repos.create(&repo, None).unwrap();

    db::repos::create(&app.db, "andy", "demo", "", hub_api::Visibility::Public).await.unwrap();

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

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

    let mut client = Repo::init(work.path()).unwrap();

    client.create_identity("Andy McCall", "[email protected]", oxy_core::repo::IdentityScope::User).unwrap();

    client.config.set("remote.origin.url", &format!("{}/andy/demo", h.base)).unwrap();

    client.config.set("remote.origin.token", "site").unwrap();

    client.save_config().unwrap();

    write(work.path(), "a.txt", b"one\n");

    client.snapshot("t").unwrap();

    let base = client.commit("base", &[]).unwrap();

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

    client.bookmark_create("feature", Some(&[base])).unwrap();

    client.bookmark_switch("feature", false).unwrap();

    write(work.path(), "b.txt", b"b\n");

    client.snapshot("t").unwrap();

    client.commit("the fix", &[]).unwrap();

    assert!(client.push_bookmark("origin", "feature", false).unwrap().iter().all(|r| r.ok));

    let preview = get(app, "/andy/demo/proposals/new?source=feature&target=main", Some(&andy)).await;

    let r = post(

        app,

        "/andy/demo/proposals/new",

        Some(&andy),

        &format!("_csrf={}&title=Fix&body=&target=main&source=feature", csrf_of(&preview.body)),

    )

    .await;

    assert_eq!(r.status, StatusCode::SEE_OTHER, "{}", r.body);

    // Runs as the queue holds them: the source's push runs (one queued,

    // one running), the proposal's own, and another bookmark's.

    let run = |bookmark: &'static str, trigger: &'static str| {

        let app = app.clone();

        let repo_id = row.id;

        async move {

            let job = db::flow::NewJob {

                name: "test".into(),

                depth: 0,

                depends: vec![],

                spec: "{}".into(),

                sandbox: "{}".into(),

                runtime: "process".into(),

                need_disk_mib: None,

            };

            let new =

                db::flow::NewRun { repo_id, bookmark, state: "s", heads: "h", pipeline_hash: "", actor: "andy", trigger, error: None };

            db::flow::create_run(&app.db, new, &[job]).await.unwrap()

        }

    };

    let queued = run("feature", "push").await;

    let running = run("feature", "push").await;

    let own = run("proposal/1", "proposal").await;

    let other = run("other", "push").await;

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

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

    let page = get(app, "/andy/demo/proposal/1", Some(&andy)).await;

    let r = act(app, "/andy/demo/proposal/1/land", Some(&andy), &format!("_csrf={}", csrf_of(&page.body))).await;

    assert_eq!(db::proposals::get(&app.db, row.id, 1).await.unwrap().unwrap().state, "landed", "{}", r.body);

    let status = |id: i64| {

        let app = app.clone();

        async move {

            let run = db::flow::run_by_id(&app.db, id).await.unwrap().unwrap();

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

            (run.status, jobs[0].error.clone())

        }

    };

    for id in [queued, own] {

        let (st, why) = status(id).await;

        assert_eq!(st, "cancelled");

        assert_eq!(why, "moot: proposal #1 landed");

    }

    assert_eq!(status(running).await.0, "running", "a started run finishes");

    assert_eq!(status(other).await.0, "queued", "another bookmark's run is untouched");

}