Cancel runs from oh, the run page, and newer pushes 745d225de344

-- Cancelled runs and jobs (oxygenflow#32). SQLite cannot change a CHECK

-- constraint in place, so both tables are rebuilt. Migrations run with

-- foreign keys off (db::migrate), or dropping flow_jobs would take its

-- logs and artifacts with it.

CREATE TABLE flow_runs_new (

    id            INTEGER PRIMARY KEY,

    repo_id       INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    bookmark      TEXT NOT NULL,

    -- The bookmark's state hash (checks are keyed by it) and its heads, joined with '+'.

    state         TEXT NOT NULL,

    heads         TEXT NOT NULL,

    pipeline_hash TEXT NOT NULL DEFAULT '',

    status        TEXT NOT NULL CHECK (status IN ('queued', 'running', 'passed', 'failed', 'error', 'cancelled')),

    -- Pipeline problems, one per line, for a run in error.

    error         TEXT NOT NULL DEFAULT '',

    actor         TEXT NOT NULL DEFAULT '',

    created_ms    INTEGER NOT NULL,

    finished_ms   INTEGER,

    number        INTEGER NOT NULL DEFAULT 0,

    trigger       TEXT NOT NULL DEFAULT 'push'

);

INSERT INTO flow_runs_new (id, repo_id, bookmark, state, heads, pipeline_hash, status, error, actor, created_ms, finished_ms, number, trigger)

    SELECT id, repo_id, bookmark, state, heads, pipeline_hash, status, error, actor, created_ms, finished_ms, number, trigger FROM flow_runs;

DROP TABLE flow_runs;

ALTER TABLE flow_runs_new RENAME TO flow_runs;

CREATE INDEX flow_runs_by_repo ON flow_runs (repo_id, id DESC);

CREATE UNIQUE INDEX flow_runs_number ON flow_runs (repo_id, number);

CREATE TABLE flow_jobs_new (

    id             INTEGER PRIMARY KEY,

    run_id         INTEGER NOT NULL REFERENCES flow_runs(id) ON DELETE CASCADE,

    name           TEXT NOT NULL,

    depth          INTEGER NOT NULL,

    -- JSON: the job's dependencies, compiled form and sandbox (flow-core types).

    depends        TEXT NOT NULL,

    spec           TEXT NOT NULL,

    sandbox        TEXT NOT NULL,

    runtime        TEXT NOT NULL,

    status         TEXT NOT NULL CHECK (status IN ('queued', 'running', 'passed', 'failed', 'skipped', 'error', 'cancelled')),

    runner         TEXT NOT NULL DEFAULT '',

    token_hash     TEXT NOT NULL DEFAULT '',

    started_ms     INTEGER,

    finished_ms    INTEGER,

    exit_code      INTEGER,

    error          TEXT NOT NULL DEFAULT '',

    receipt        TEXT NOT NULL DEFAULT '',

    receipt_hash   TEXT NOT NULL DEFAULT '',

    log_count      INTEGER NOT NULL DEFAULT 0,

    also_read      TEXT NOT NULL DEFAULT '',

    receipt_attest TEXT NOT NULL DEFAULT '',

    receipt_signer TEXT NOT NULL DEFAULT '',

    receipt_note   TEXT NOT NULL DEFAULT '',

    runner_id      INTEGER,

    timeout_s      INTEGER NOT NULL DEFAULT 0,

    UNIQUE (run_id, name)

);

INSERT INTO flow_jobs_new (id, run_id, name, depth, depends, spec, sandbox, runtime, status, runner, token_hash, started_ms, finished_ms,

        exit_code, error, receipt, receipt_hash, log_count, also_read, receipt_attest, receipt_signer, receipt_note, runner_id, timeout_s)

    SELECT id, run_id, name, depth, depends, spec, sandbox, runtime, status, runner, token_hash, started_ms, finished_ms,

        exit_code, error, receipt, receipt_hash, log_count, also_read, receipt_attest, receipt_signer, receipt_note, runner_id, timeout_s FROM flow_jobs;

DROP TABLE flow_jobs;

ALTER TABLE flow_jobs_new RENAME TO flow_jobs;

CREATE INDEX flow_jobs_queued ON flow_jobs (status, id);

CREATE INDEX flow_jobs_token ON flow_jobs (token_hash) WHERE token_hash <> '';

CREATE INDEX flow_jobs_runner ON flow_jobs (runner_id);

    // The migrations `sqlx::migrate!` embeds.

    println!("cargo:rerun-if-changed=migrations");

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

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

/// with the run as it now stands.

async fn cancel_flow_run(

    State(app): State<Shared>,

    headers: HeaderMap,

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

) -> Result<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()))?;

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

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

    Ok(Json(serde_json::json!({ "number": number, "status": "cancelled" })))

}

        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error" | "cancelled")

/// The run a push to `bookmark` is scoped against: the newest that was not

/// cancelled, nor a push run still going (a new push supersedes it), so

/// that what a cancelled run would have checked is checked by the next.

    sqlx::query_as(

        "SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND heads != '' AND status != 'cancelled'

         AND NOT (trigger = 'push' AND status IN ('queued', 'running')) ORDER BY id DESC LIMIT 1",

    )

    .bind(repo_id)

    .bind(bookmark)

    .fetch_optional(pool)

    .await

        if let Some(x) = job(pool, id).await? {

            out.push(x);

        }

    }

    Ok(out)

}

/// Cancel a run (oxygenflow#32): its jobs not yet done end as cancelled,

/// and so does the run. A runner learns of it when the hub next refuses

/// its log lines, and stops the job. Returns the jobs whose status changed.

pub async fn cancel_run(pool: &Pool, run_id: i64, why: &str) -> sqlx::Result<Vec<Job>> {

    let now = now_ms();

    let mut tx = pool.begin().await?;

    let ids: Vec<(i64,)> = sqlx::query_as("SELECT id FROM flow_jobs WHERE run_id = ? AND status IN ('queued', 'running')")

        .bind(run_id)

        .fetch_all(&mut *tx)

        .await?;

    sqlx::query(

        "UPDATE flow_jobs SET status = 'cancelled', error = ?, finished_ms = ?, token_hash = ''

         WHERE run_id = ? AND status IN ('queued', 'running')",

    )

    .bind(why)

    .bind(now)

    .bind(run_id)

    .execute(&mut *tx)

    .await?;

    sqlx::query("UPDATE flow_runs SET status = 'cancelled', finished_ms = ? WHERE id = ? AND status IN ('queued', 'running')")

        .bind(now)

        .bind(run_id)

        .execute(&mut *tx)

        .await?;

    tx.commit().await?;

    let mut out = Vec::new();

    for (id,) in ids {

}

/// The push runs of `bookmark` still going, older than run `number`.

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

    sqlx::query_as(

        "SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND number < ? AND trigger = 'push'

         AND status IN ('queued', 'running') ORDER BY id",

    )

    .bind(repo_id)

    .bind(bookmark)

    .bind(number)

    .fetch_all(pool)

    .await

    migrate(options.clone()).await?;

/// Apply the migrations on a connection of their own with foreign keys

/// off, as SQLite's way to change a table (rebuild it, drop the old one,

/// rename) needs: with them on, dropping a table its children reference

/// deletes the children too. Every reference must hold afterwards.

async fn migrate(options: SqliteConnectOptions) -> anyhow::Result<()> {

    use sqlx::Connection;

    let mut conn = sqlx::SqliteConnection::connect_with(&options.foreign_keys(false)).await?;

    sqlx::migrate!("./migrations").run(&mut conn).await?;

    let broken: Vec<(String,)> = sqlx::query_as("SELECT \"table\" FROM pragma_foreign_key_check").fetch_all(&mut conn).await?;

    if let Some((table,)) = broken.first() {

        anyhow::bail!("after the migrations, {} rows break a foreign key (the first in {table})", broken.len());

    }

    conn.close().await?;

    Ok(())

}

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

            if let Ok(Some(n)) = started {

                supersede(app, row, bookmark, n).await;

            }

            started

}

/// Cancel a run still going (oxygenflow#32), its checks with it.

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

    if !run.live() {

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

    }

    cancel_with(app, run, &format!("cancelled by {actor}")).await.map_err(|e| {

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

        "the run could not be cancelled".to_string()

    })

}

async fn cancel_with(app: &Shared, run: &fdb::Run, why: &str) -> sqlx::Result<()> {

    let changed = fdb::cancel_run(&app.db, run.id, why).await?;

    report_changed(app, run.id, changed).await

/// A push makes the push runs still going on its bookmark moot: they are

/// cancelled, and the new run's scope reaches back past them.

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

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

    for run in older {

        if let Err(e) = cancel_with(app, &run, &format!("superseded by run #{number}")).await {

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

        }

    }

}

        "cancelled" => "cancelled",

    // An empty batch asks only whether the job still runs: a cancelled

    // one answers 409 and its runner stops it (oxygenflow#32).

    if batch.records.is_empty() {

        return Ok(StatusCode::NO_CONTENT);

    }

            "cancelled" => j.error.clone(),

/// `POST /{owner}/{name}/flow/{number}/cancel` (oxygenflow#32).

pub async fn cancel(

    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::cancel(&app, &row, &run, &me.account.username).await {

        Ok(()) => {

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

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

        }

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

    })

}

        "cancelled" => "⊘",

    /// Whether the viewer may cancel it.

    can_cancel: bool,

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

    let can_rerun = !run.live() && can_write && crate::flow::enabled(&app, &row);

    let can_cancel = run.live() && can_write;

        can_cancel,

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

        // Passed, so the next push is scoped against it, not superseding it.

        sqlx::query("UPDATE flow_runs SET status = 'passed' WHERE id = ?").bind(runs[0].id).execute(&app.db).await.unwrap();

/// oxygenflow#32: a run still going is cancelled from the API, the web or

/// by a newer push to its bookmark; its runner is refused from then on,

/// and the next push is scoped past the cancelled run.

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

async fn runs_are_cancelled_by_hand_or_by_a_newer_push() {

    let h = hub().await;

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

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

    let url = format!("{}/andy/stop", h.base);

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

    let c = client(&url, work.path());

    let extra = "job \"dist\" {\n    sandbox \"alpine\"\n    scope \"Cargo.toml\"\n    exec \"true\"\n}\n";

    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));

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

    write(work.path(), "Cargo.toml", "version = \"0.1.0\"\n");

    c.commit("Start", &[]).unwrap();

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

    let wait_runs = |app: Shared, n: usize| async move {

        for _ in 0..100 {

            let runs = runs_of(&app, "andy", "stop").await;

            if runs.len() >= n {

                return runs;

            }

            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        }

        panic!("no run {n}");

    };

    let first = wait_runs(h.app.clone(), 1).await;

    sqlx::query("UPDATE flow_runs SET status = 'passed' WHERE id = ?").bind(first[0].id).execute(&h.app.db).await.unwrap();

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

    // Run 2 (a version bump) has a job running when run 3 is pushed.

    write(work.path(), "Cargo.toml", "version = \"0.1.1\"\n");

    c.commit("Release 0.1.1", &[]).unwrap();

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

    wait_runs(h.app.clone(), 2).await;

    let job = claim(&h.app).await.expect("a job of run 2");

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

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

    c.commit("Edit the test", &[]).unwrap();

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

    let runs = wait_runs(h.app.clone(), 3).await;

    let third = &runs[0];

    let mut second = runs[1].clone();

    for _ in 0..100 {

        second = db::flow::run_by_id(&h.app.db, runs[1].id).await.unwrap().unwrap();

        if !second.live() {

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    assert_eq!(second.status, "cancelled", "a newer push supersedes a run still going");

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

    assert!(jobs.iter().all(|j| j.status == "cancelled" && j.error == "superseded by run #3"), "{jobs:?}");

    let mut names: Vec<String> = db::flow::jobs(&h.app.db, third.id).await.unwrap().into_iter().map(|j| j.name).collect();

    names.sort();

    assert_eq!(names, ["build", "dist", "lint", "test"], "run 3 also checks what run 2 would have: the version bump");

    let mut checks = vec![];

    for _ in 0..100 {

        checks = db::hooks::checks_for_state(&h.app.db, second.repo_id, &second.state).await.unwrap();

        if checks.iter().all(|c| c.status == "cancelled") {

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    assert!(checks.iter().all(|c| c.status == "cancelled"), "{checks:?}");

    // The runner of the cancelled job is refused, so it stops the job.

    let (st, _) =

        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(r#"{"records":[]}"#.into())).await;

    assert_eq!(st, StatusCode::CONFLICT);

    // By hand: the API, then a run already ended is refused.

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

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

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

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

    assert!(db::flow::jobs(&h.app.db, third.id).await.unwrap().iter().all(|j| j.error == "cancelled by andy"));

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

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

    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", None, None).await;

    assert_ne!(st, StatusCode::OK, "anonymous callers cannot cancel");

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

    assert!(page.contains("⊘") && !page.contains("CANCEL</button>"), "a cancelled run shows so, with no button");

}

    /// Cancel a run still queued or running; its running jobs are stopped

    Cancel { number: i64 },

        RunCmd::Cancel { number } => {

            let done: serde_json::Value = c.send("POST", &format!("{base}/flow/runs/{number}/cancel"), &serde_json::json!({}))?;

            if json {

                println!("{done}");

            } else {

                println!("{} cancelled run #{number}", ui::ok("✔"));

            }

        }

| `oh run cancel N` | Cancel a run still queued or running; its running jobs are stopped |

JSON under `/api/v1`, bearer tokens, `application/json`, RFC 7807 problem responses (an unknown route too), 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` (`?state=open|draft|landed|closed|all`, open ones and drafts by default; a source bookmark may hold `/`, and `owner/name:bookmark` naming the repository itself is its own bookmark), `/proposals/{n}/comments|reviews|checks|files`, `PATCH /proposals/{n}` (the title and description, for its author or a maintainer, and `labels`, the whole set by name, for triage; what is not given stays), `POST /proposals/{n}/land` (repo:write: what the Land button does, `409` with the blockers when it cannot; a landed proposal keeps the changes it landed and reports `landed` with who, how (`web`, or `api (token NAME)`) and when, which its event, the audit log and the push record also say, adding when nobody approved), `/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` (`PATCH /releases/{tag}` changes the title, notes or prerelease flag, repo:write), `/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/{state}` (the checks on a state, named by its hash, by its heads joined with `+`, by the one head of a single-head state, or by a bookmark), `/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` (`GET`: runs newest first, `?before=N&limit=`; `POST {"bookmark"}` or `{"change"}`, repo:write: run a bookmark's current state, the default bookmark when none is named, or one change's state), `/flow/runs/{n}` (a run with its jobs), `/flow/runs/{n}/jobs/{job}/log` (`?after=seq&limit=`) and `/flow/runs/{n}/rerun` (repo:write: a finished run again), both answering `201 {"number", "url"}` or `409` with why nothing started, `/flow/runs/{n}/cancel` (repo:write: a run still going ends as `cancelled`, its running jobs stopped when their runner next reports; `409` when it had already ended; a push to a bookmark cancels the push runs of it still going), `/search`. A change's author carries the account its key is linked to, or `null` and a `note` saying the key is not linked and where its owner links it (the change page shows a "key not linked" tag, with a link to Settings → Keys for signed-in viewers). 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.

  {% if can_cancel %}<form class="inline" method="post" action="{{ repo.url }}/flow/{{ run.number }}/cancel" onsubmit="return confirm('Cancel run #{{ run.number }}? Jobs still running are stopped.')"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="small danger-btn">CANCEL</button></form>{% endif %}

.flow-glyph.st-queued, .flow-glyph.st-skipped, .flow-glyph.st-cancelled, .node.st-queued .g, .node.st-skipped .g, .node.st-cancelled .g { color: var(--mute); fill: var(--mute); }

.flow-graph .node.st-skipped rect, .flow-graph .node.st-cancelled rect { stroke-dasharray: 4 3; }