Rerun one job of a run, with what depends on it e8fa53694f68

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

/// `POST /repos/{o}/{n}/flow/runs/{number}/jobs/{job}/rerun` (repo:write,

/// oxygenflow#31): that job again, with the jobs that depend on it, in the

/// same run. Answers with how many jobs were queued.

async fn rerun_flow_job(

    State(app): State<Shared>,

    headers: HeaderMap,

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

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

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

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

    Ok(Json(serde_json::json!({ "run": number, "job": job, "queued": n })))

}

/// Run job `name` of finished run `run_id` again, with every job that

/// depends on it (oxygenflow#31): back to the queue as if new, their logs,

/// artifacts and reports gone, and the run running again. The jobs they

/// depend on keep their results. The jobs reset; none when there is no

/// such job.

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

    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(run_id).fetch_all(pool).await?;

    if !all.iter().any(|j| j.name == name) {

        return Ok(Vec::new());

    }

    let mut again = vec![name.to_string()];

    loop {

        let more: Vec<String> = all

            .iter()

            .filter(|j| !again.contains(&j.name) && j.depends_on().iter().any(|d| again.contains(d)))

            .map(|j| j.name.clone())

            .collect();

        if more.is_empty() {

            break;

        }

        again.extend(more);

    }

    let ids: Vec<i64> = all.iter().filter(|j| again.contains(&j.name)).map(|j| j.id).collect();

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

    for id in &ids {

        for table in ["flow_logs", "flow_artifacts", "flow_declines", "test_reports", "coverage_reports", "reports"] {

            sqlx::query(&format!("DELETE FROM {table} WHERE job_id = ?")).bind(id).execute(&mut *tx).await?;

        }

        sqlx::query(

            "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, finished_ms = NULL,

             exit_code = NULL, error = '', receipt = '', receipt_hash = '', log_count = 0, also_read = '', receipt_attest = '',

             receipt_signer = '', receipt_note = '', runner_id = NULL, waiting = '', placed = '', peak_disk_mib = NULL,

             peak_memory_mib = NULL WHERE id = ?",

        )

        .bind(id)

        .execute(&mut *tx)

        .await?;

    }

    sqlx::query("UPDATE flow_runs SET status = 'running', finished_ms = NULL WHERE id = ?").bind(run_id).execute(&mut *tx).await?;

    tx.commit().await?;

    let mut out = Vec::new();

    for id in ids {

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

            out.push(x);

        }

    }

    Ok(out)

}

/// Run one job of a finished run again (oxygenflow#31), with the jobs

/// that depend on it; the jobs it depends on keep their results, so a

/// flaky test reruns without the build before it. Its checks go back to

/// queued. Returns how many jobs were queued again.

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

    if !enabled(app, row) {

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

    }

    if run.live() {

        return Err(format!("run #{} is still {}: wait for it, or cancel it, first", run.number, run.status));

    }

    let changed = fdb::rerun_job(&app.db, run.id, job).await.map_err(|e| {

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

        "the job could not be queued again".to_string()

    })?;

    if changed.is_empty() {

        return Err(format!("run #{} has no job {job:?}", run.number));

    }

    for j in &changed {

        report_check(app, row, &run.state, run.number, &j.name, "queued", &format!("queued again by {actor}")).await;

    }

    Ok(changed.len())

}

/// `POST /{owner}/{name}/flow/{number}/jobs/{job}/rerun` (oxygenflow#31).

pub async fn rerun_job(

    State(app): State<Shared>,

    viewer: Viewer,

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

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

        Ok(_) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("{job} of #{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(),

    })

}

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

    let opts = RemoteOptions { token: Some(token.clone()), ..Default::default() };

    let bad = RemoteOptions { token: Some("flowjob_00".into()), ..Default::default() };

/// oxygenflow#31: one finished job runs again in the same run, with the

/// jobs that depend on it; what it depends on keeps its result, so a flaky

/// test reruns without its build. Not while the run is live, and only a

/// job the run has.

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

async fn one_job_reruns_with_what_depends_on_it() {

    let h = hub().await;

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

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

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

    let c = client(&format!("{}/andy/flaky", h.base), 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 run = runs_of(&h.app, "andy", "flaky").await[0].clone();

    // build and lint pass; test fails, as a flaky one does.

    let (st, _) =

        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/test/rerun", run.number), Some("site"), None).await;

    assert_eq!(st, StatusCode::CONFLICT, "not while the run is live");

    while let Some(j) = claim(&h.app).await {

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

        let (name, ph) = (j["job"]["name"].as_str().unwrap().to_string(), j["pipeline_hash"].as_str().unwrap().to_string());

        let exit = if name == "test" { 1 } else { 0 };

        let log = serde_json::json!({ "records": [{ "t": 1, "stream": "out", "text": format!("{name} ran") }] });

        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(log.to_string())).await;

        assert_eq!(

            finish(&h.app, id, serde_json::json!({ "exit_code": exit, "duration_ms": 10, "receipt": receipt(&name, &ph, exit as u32) }))

                .await,

            StatusCode::NO_CONTENT

        );

    }

    let status = |n: &'static str| {

        let (app, run_id) = (h.app.clone(), run.id);

        async move { db::flow::jobs(&app.db, run_id).await.unwrap().into_iter().find(|j| j.name == n).unwrap() }

    };

    assert_eq!(db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap().status, "failed");

    let failed_test = status("test").await;

    // The test again: build and lint keep their results.

    let (st, body) =

        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/test/rerun", run.number), Some("site"), None).await;

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

    assert!(body.contains(r#""queued":1"#), "{body}");

    assert_eq!(

        (status("build").await.status, status("lint").await.status, status("test").await.status),

        ("passed".into(), "passed".into(), "queued".into())

    );

    assert!(db::flow::log(&h.app.db, failed_test.id, 0, 100).await.unwrap().is_empty(), "its old log is gone");

    assert!(db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap().live());

    let checks = checks_when(&h.app, run.repo_id, &run.state, 3).await;

    assert_eq!(checks.iter().find(|c| c.name == "flow/test").map(|c| c.status.as_str()), Some("queued"));

    let j = claim(&h.app).await.expect("the test, queued again");

    assert_eq!(j["job"]["name"], "test");

    let ph = j["pipeline_hash"].as_str().unwrap().to_string();

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

    assert_eq!(

        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt("test", &ph, 0) })).await,

        StatusCode::NO_CONTENT

    );

    assert_eq!(db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap().status, "passed", "the run is whole again");

    // The build again brings the test that depends on it; lint stays.

    let (_, body) =

        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/build/rerun", run.number), Some("site"), None).await;

    assert!(body.contains(r#""queued":2"#), "{body}");

    assert_eq!(

        (status("build").await.status, status("lint").await.status, status("test").await.status),

        ("queued".into(), "passed".into(), "queued".into())

    );

    let (st, _) =

        call(&h.app, "POST", &format!("/api/v1/repos/andy/flaky/flow/runs/{}/jobs/deploy/rerun", run.number), Some("site"), None).await;

    assert_eq!(st, StatusCode::CONFLICT, "live again, and no such job either way");

}

    /// Run a finished run's state again; with --job, only that job (and the

    /// jobs that depend on it) in the same run

    Rerun {

        number: i64,

        #[arg(long)]

        job: Option<String>,

    },

            }

        }

        RunCmd::Rerun { number, job: Some(job) } => {

            let queued: serde_json::Value =

                c.send("POST", &format!("{base}/flow/runs/{number}/jobs/{job}/rerun"), &serde_json::json!({}))?;

            if json {

                println!("{queued}");

            } else {

                let n = queued["queued"].as_i64().unwrap_or(1);

                let with = if n > 1 { format!(" with the {} that depend on it", n - 1) } else { String::new() };

                println!("{} queued {job} of run #{number} again{with}", ui::ok("✔"));

        RunCmd::Rerun { number, job: None } => {

JSON under `/api/v1`, bearer tokens, `application/json`, RFC 7807 problem responses (an unknown route, or a known one with the wrong method, 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/{ref}` (`?path=`, `?offset=&limit=`, or `?r=REVSET` for the changes a revset selects, evaluated under a two-second budget and answering 400 when it does not parse or goes past it; Oxygen's docs/revset_query_language.md), `/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` (`POST` creates one or changes the description and due date of the one with that title, `PATCH /milestones/{title}` changes title, description, due date or state; triage), `/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, and `owner` to transfer it to an owner the caller may make boards under (#125: it takes the next number there, and its old address redirects), `/cards` to list them by column or 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), `/coverage` and `/tests` (a bookmark's coverage and JUnit test reports from Flow jobs, newest first, `?bookmark=&limit=`), `POST /coverage` (repo:write, #136: `{"bookmark"}` or `{"change"}` and `content`, an LCOV or Cobertura report measured off the hub, kept for that state as uploaded by the caller, shown "uploaded by NAME" wherever a run number would be, and setting no check; `oh coverage upload FILE`), `/coverage/{id}` (a report with its files) and `/tests/{id}` (a report with its cases, `?status=failed` for the failed and errored ones), `/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; with `"jobs": [NAME..]` only those jobs and what they depend on, each one a manual run starts, else `409` naming the ones it does (#130); `oh run start --job NAME`), `/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}/jobs/{job}/rerun` (repo:write, oxygenflow#31: that job of a finished run again, in the same run, with the jobs that depend on it; what it depends on keeps its result, so a flaky test reruns without its build; their logs, artifacts and reports go, their checks return to queued and the run is running again; `{"queued": N}`, or `409` while the run is live or for a job it lacks; RE-RUN THIS JOB on the run page, `oh run rerun N --job NAME`), `/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; `via` names the delegate (an agent) that signed for the author when one did, with the delegation's scope and expiry and whether it has since been revoked, and the pages show the same as an `agent` badge, a `via` row under the author on the change page, and a count under the author in Contributors (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_rerun && l.j.done() %}<form class="tiny inline job-rerun" method="post" action="{{ repo.url }}/flow/{{ run.number }}/jobs/{{ l.j.name }}/rerun"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="small" title="Run {{ l.j.name }} again in this run, with the jobs that depend on it; what it depends on keeps its results">RE-RUN THIS JOB</button></form>{% endif %}