A push reuses a job's pass on the same state (closes #265) 5588bb3631d2

-- A push run reuses a job's pass on the same state, spec and sandbox

-- instead of running it again (oxygenhub#265); a maintainer can turn

-- that off for the repository.

ALTER TABLE repositories ADD COLUMN flow_reuse BOOLEAN NOT NULL DEFAULT TRUE;

-- A push run reuses a job's pass on the same state, spec and sandbox

-- instead of running it again (oxygenhub#265); a maintainer can turn

-- that off for the repository.

ALTER TABLE repositories ADD COLUMN flow_reuse INTEGER NOT NULL DEFAULT 1;

- **A push reuses a job's pass on the same state (#265):** when main reaches a state a proposal's run already tested (landing it), a job that passed there with the same spec and sandbox is not run again: it shows passed, reused from that run, with its receipt, and its check says so. Jobs with an approval gate, an environment, secrets or identity tokens, or whose artifacts another job takes, always run. Maintainers can turn reuse off under Settings › Runs.

    /// The run whose pass this job reused (#265), if it did.

    pub fn reused_from(&self) -> Option<i64> {

        self.error.strip_prefix("reused from run #")?.parse().ok()

    }

    /// A pass on the same state to reuse instead of running it (#265): the

    /// job is created passed, with that pass's receipt.

    pub reused: Option<Reused>,

}

/// A job's earlier pass on the same state, spec and sandbox (#265).

#[derive(Clone, Debug)]

pub struct Reused {

    /// The run it passed in.

    pub run_number: i64,

    pub receipt: String,

    pub receipt_hash: String,

    pub receipt_attest: String,

    pub receipt_signer: String,

    pub receipt_note: String,

}

impl Reused {

    /// What the reused job says, and how [`Job::reused_from`] reads it back.

    pub fn note(&self) -> String {

        format!("reused from run #{}", self.run_number)

    }

}

/// The newest pass of job `name` with exactly this spec and sandbox on

/// `state` in this repository (#265).

pub async fn passed_on_state(

    pool: &Pool,

    repo_id: i64,

    state: &str,

    name: &str,

    spec: &str,

    sandbox: &str,

) -> sqlx::Result<Option<Reused>> {

    let row: Option<(i64, String, String, String, String, String)> = sqlx::query_as(

        "SELECT r.number, j.receipt, j.receipt_hash, j.receipt_attest, j.receipt_signer, j.receipt_note \

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \

         WHERE r.repo_id = $1 AND r.state = $2 AND j.name = $3 AND j.spec = $4 AND j.sandbox = $5 AND j.status = 'passed' \

         ORDER BY j.id DESC LIMIT 1",

    )

    .bind(repo_id)

    .bind(state)

    .bind(name)

    .bind(spec)

    .bind(sandbox)

    .fetch_optional(pool)

    .await?;

    Ok(row.map(|(run_number, receipt, receipt_hash, receipt_attest, receipt_signer, receipt_note)| Reused {

        run_number,

        receipt,

        receipt_hash,

        receipt_attest,

        receipt_signer,

        receipt_note,

    }))

        let job_id: i64 = sqlx::query_scalar(

             VALUES ($1, $2, $3, $4, $5, $6, $7, 'queued', $8, $9, $10, $11, $12) RETURNING id",

        .fetch_one(&mut *tx)

        // A pass reused (#265): done before any runner can claim it.

        if let Some(r) = &j.reused {

            sqlx::query(

                "UPDATE flow_jobs SET status = 'passed', exit_code = 0, error = $1, receipt = $2, receipt_hash = $3, receipt_attest = $4, \

                 receipt_signer = $5, receipt_note = $6, started_ms = $7, finished_ms = $7 WHERE id = $8",

            )

            .bind(r.note())

            .bind(&r.receipt)

            .bind(&r.receipt_hash)

            .bind(&r.receipt_attest)

            .bind(&r.receipt_signer)

            .bind(&r.receipt_note)

            .bind(now)

            .bind(job_id)

            .execute(&mut *tx)

            .await?;

        }

    }

    // Every job reused: the run has passed already.

    if !jobs.is_empty() && jobs.iter().all(|j| j.reused.is_some()) {

        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = $1 WHERE id = $2").bind(now).bind(id).execute(&mut *tx).await?;

    /// A push run reuses a job's pass on the same state (#265, migration

    /// 0090). On by default.

    #[sqlx(default)]

    pub flow_reuse: bool,

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

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

    Ok(())

}

/// Whether a job's earlier pass may stand for it on the same state (#265):

/// not a job that waits for a person, deploys to an environment, holds

/// secrets or asks for identity tokens, whose running is the point.

fn reusable(job: &flow_core::Job) -> bool {

    job.approval.is_none() && job.environment.is_none() && job.secrets.is_empty() && job.identities.is_empty()

}

                        reused: None,

            }

            // A job that passed on this very state, with the same spec and

            // sandbox, is not run again by a push (#265): main after a

            // proposal landed it as tested. Never one that publishes, deploys

            // or holds secrets, nor one whose artifacts a job of this run

            // takes; and not where the repository turned reuse off.

            if trigger == "push" && row.flow_reuse {

                let consumed: std::collections::HashSet<&str> = p

                    .order()

                    .into_iter()

                    .filter(|j| keep.contains(&j.name))

                    .flat_map(|j| j.inputs.iter().map(|i| i.from.as_str()))

                    .collect();

                for j in &mut jobs {

                    let Some(spec) = p.jobs.get(&j.name) else { continue };

                    if !reusable(spec) || consumed.contains(j.name.as_str()) {

                        continue;

                    }

                    j.reused =

                        fdb::passed_on_state(&app.db, row.id, state, &j.name, &j.spec, &j.sandbox).await.map_err(|e| e.to_string())?;

                }

                for j in jobs.iter().filter(|j| j.reused.is_none()) {

                // Reused passes (#265) report as any pass does: their checks,

                // and a run that passed with them (Diffusion, landing a bump).

                if jobs.iter().any(|j| j.reused.is_some()) {

                    let reused: Vec<fdb::Job> =

                        fdb::jobs(&app.db, run.id).await.unwrap_or_default().into_iter().filter(|j| j.reused_from().is_some()).collect();

                    if let Err(e) = report_changed(app, run.id, reused).await {

                        tracing::warn!("flow: reporting reused passes of run #{}: {e}", run.number);

                    }

                }

            "passed" => match j.reused_from() {

                Some(n) => format!("passed in run #{n} on this state, reused"),

                None => format!("passed in {}", duration(&j)),

            },

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

pub struct ReuseForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `on` when the box is ticked.

    #[serde(default)]

    reuse: String,

}

/// `POST /{owner}/{repo}/settings/flow-reuse` (oxygenhub#265): whether a

/// push run reuses a job's pass on the same state; for maintainers,

/// audited.

pub async fn flow_reuse(

    State(app): State<Shared>,

    viewer: Viewer,

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

    Form(f): Form<ReuseForm>,

) -> Result<Response, Error> {

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

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

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

    let on = f.reuse == "on";

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

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

    let notice = if on {

        "A push run now reuses a job's pass on the same state."

    } else {

        "Every job of every push now runs, passed before or not."

    };

    back(&row, Section::Runs, Some(notice.into()), None)

}

#[derive(Deserialize)]

/// oxygenhub#265: a push run on a state whose jobs already passed reuses

/// those passes (same spec and sandbox), except a job behind a gate; a

/// repository can turn reuse off.

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

async fn a_push_reuses_a_pass_on_the_same_state() {

    let h = hub().await;

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

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

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

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

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

    let gated = "job \"deploy\" {\n    sandbox \"alpine\"\n    depends-on \"test\"\n    approval {\n        by \"maintain\"\n    }\n    exec \"true\"\n}\n";

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

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

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

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

    let first = runs_of(&h.app, "andy", "reuse").await.remove(0);

    // Its jobs pass, the gated one too, each with a receipt.

    sqlx::query("UPDATE flow_jobs SET status = 'passed', exit_code = 0, receipt_hash = 'r-' || name WHERE run_id = $1")

        .bind(first.id)

        .execute(&h.app.db)

        .await

        .unwrap();

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

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

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

    assert!(row.flow_reuse, "on by default");

    let again = |row: db::repos::RepoRow| {

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

        async move {

            let n =

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

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

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

            (run, jobs)

        }

    };

    let (run, jobs) = again(row.clone()).await;

    let by = |name: &str| jobs.iter().find(|j| j.name == name).unwrap().clone();

    for name in ["build", "test", "lint"] {

        let j = by(name);

        assert_eq!((j.status.as_str(), j.reused_from()), ("passed", Some(first.number)), "{name}: {j:?}");

        assert_eq!(j.receipt_hash, format!("r-{name}"), "the pass's receipt stands for it");

    }

    let deploy = by("deploy");

    assert!(deploy.status == "queued" && deploy.reused_from().is_none(), "a gated job runs again: {deploy:?}");

    assert_eq!(run.status, "queued", "the run waits for the gated job");

    let (_, page) = call(&h.app, "GET", &format!("/andy/reuse/flow/{}", run.number), None, None).await;

    assert!(page.contains("data-reused") && page.contains(&format!("/andy/reuse/flow/{}", first.number)), "{page}");

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

    assert!(

        checks.iter().any(|c| c.name == "flow/build" && c.status == "passed" && c.summary.contains("reused")),

        "a reused pass reports its check: {checks:?}"

    );

    // Turned off: every job runs.

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

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

    let (_, jobs) = again(row).await;

    assert!(jobs.iter().all(|j| j.status == "queued" && j.reused_from().is_none()), "{jobs:?}");

}

                reused: None,

- **Reusing a pass** (oxygenhub#265). A push run does not run a job again when the same job, with the same compiled spec and sandbox (the image pin included), already passed on the very same state in this repository, as main does after a proposal lands its tested projection: the job is created passed, its `error` note `reused from run #N`, carrying that pass's receipt, hash and attestation, and its check reports `passed in run #N on this state, reused`. A run whose jobs are all reused has passed when created, and reports as any passing run (Diffusion's green bumps, a bump's landing). Never reused: a job with an approval gate, an environment, secrets or identity tokens, or one whose artifacts another job of the run takes as inputs; manual runs, re-runs, schedules and tags always run. Maintainers turn it off under Settings › Runs (`repositories.flow_reuse`, audited as `repo.flow_reuse`).

  {% if l.j.status == "passed" %}{% if let Some(n) = l.j.reused_from() %}<p class="small mute pad" data-reused>Not run again: this job, with the same spec and sandbox, passed on this very state in <a href="{{ repo.url }}/flow/{{ n }}">run #{{ n }}</a>, whose log it would repeat; that pass's receipt stands for it here.</p>{% endif %}{% endif %}

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

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

  <h2>Reuse passes</h2>

  <p>A push run doesn't run a job again when the same job, with the same spec and sandbox, already passed on the very same state, as main does after a proposal lands: the job shows as passed, reused from that run, with its receipt. A job that waits for an approval, deploys to an environment, holds secrets or asks for identity tokens always runs, as does one whose artifacts another job of the run takes.</p>

  <label class="check"><input type="checkbox" name="reuse"{% if row.flow_reuse %} checked{% endif %}> Reuse a job's pass on the same state (untick to run every job of every push, for flaky or time-dependent tests)</label>

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

</form>