Runs started by a push honour each job's scope d7ce41acf22c

- **A job's `scope` is honoured (#88).** A run a push starts leaves out each job whose `scope` globs none of the files the push's new changes touch (since the bookmark's previous run), unless a job that runs depends on it; a bookmark's first run, "Run pipeline" and reruns keep every job. A push that touches no job's business starts no run.

 "globset",

# A job's scope globs (oxygen#59).

globset = "0.4"

/// The latest run on this bookmark of this repository, for what a push

/// changed since (a job's scope).

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

    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND heads != '' ORDER BY id DESC LIMIT 1")

        .bind(repo_id)

        .bind(bookmark)

        .fetch_optional(pool)

        .await

}

    let touched = touched_since_last_run(app, row, bookmark, heads).await;

    if let Err(e) = start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).await {

/// The paths the changes a push brought touch: the changes in `heads`'

/// state that the bookmark's previous run did not have, and the files their

/// edits and tree operations name, as the new state places them. `None`

/// (run every job) for the bookmark's first run, or when it cannot be told.

async fn touched_since_last_run(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Option<Vec<String>> {

    let last = fdb::latest_on(&app.db, row.id, bookmark).await.ok()??;

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

    if old.is_empty() {

        return None;

    }

    let repo = RepoName::new(&row.owner, &row.name)?;

    let new = heads.to_vec();

    app.repos

        .read(&repo, move |r, cache| {

            let fresh = r.negotiate(&old, &new)?;

            let loaded = cache.load(r, &new)?;

            let by_node: std::collections::HashMap<oxy_core::ids::NodeId, &String> =

                loaded.snapshot.files.iter().map(|(p, f)| (f.node, p)).collect();

            let mut paths = std::collections::BTreeSet::new();

            for h in fresh {

                let (c, _) = r.store().get_change_lazy(&h)?;

                let nodes = c.edits.iter().map(|e| e.node()).chain(c.tree_ops.iter().map(|o| o.node()));

                for n in nodes {

                    if let Some(p) = by_node.get(&n) {

                        paths.insert((*p).clone());

                    }

                }

            }

            Ok(paths.into_iter().collect::<Vec<String>>())

        })

        .await

        .ok()

}

/// Whether a job runs for a push that touched `touched`: always without a

/// scope or without knowing what was touched, else when a path matches.

fn in_scope(scope: &[String], touched: Option<&[String]>) -> bool {

    let Some(touched) = touched else { return true };

    if scope.is_empty() {

        return true;

    }

    let mut b = globset::GlobSetBuilder::new();

    for g in scope {

        match globset::Glob::new(g) {

            Ok(g) => {

                b.add(g);

            }

            Err(_) => return true,

        }

    }

    let Ok(set) = b.build() else { return true };

    touched.iter().any(|p| set.is_match(p))

}

/// [`start_run`] for a push: jobs whose `scope` the push did not touch are

/// left out of the run, unless a job that runs depends on them.

#[allow(clippy::too_many_arguments)]

pub async fn start_run_scoped(

    app: &Shared,

    row: &db::repos::RepoRow,

    bookmark: &str,

    state: &str,

    heads: &[ChangeHash],

    actor: &str,

    trigger: &str,

    touched: Option<&[String]>,

) -> Result<Option<i64>, String> {

    start_run_with(app, row, bookmark, state, heads, actor, trigger, touched).await

}

    start_run_with(app, row, bookmark, state, heads, actor, trigger, None).await

}

#[allow(clippy::too_many_arguments)]

async fn start_run_with(

    app: &Shared,

    row: &db::repos::RepoRow,

    bookmark: &str,

    state: &str,

    heads: &[ChangeHash],

    actor: &str,

    trigger: &str,

    touched: Option<&[String]>,

) -> Result<Option<i64>, String> {

            // Jobs out of the push's scope stay out, unless a kept job

            // needs them.

            let mut keep: std::collections::HashSet<String> =

                p.order().into_iter().filter(|j| in_scope(&j.scope, touched)).map(|j| j.name.clone()).collect();

            loop {

                let needed: Vec<String> = p

                    .order()

                    .into_iter()

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

                    .flat_map(|j| j.depends_on.iter().cloned())

                    .filter(|d| !keep.contains(d))

                    .collect();

                if needed.is_empty() {

                    break;

                }

                keep.extend(needed);

            }

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

            if jobs.is_empty() && touched.is_some() {

                // Nothing this push touched is any job's business.

                return Ok(None);

            }

/// oxygen#59: a job with a `scope` runs on a push only when the push's

/// changes touch a path it names (a release job, when Cargo.toml's version

/// moves); a bookmark's first run and manual runs have every job.

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

async fn a_scoped_job_runs_only_when_a_push_touches_its_paths() {

    let h = hub().await;

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

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

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

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

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

    let extra = "job \"dist\" {\n    sandbox \"alpine\"\n    scope \"Cargo.toml\" \"packaging/**\"\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 names = |app: Shared, n: usize| async move {

        let mut runs = Vec::new();

        for _ in 0..100 {

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

            if runs.len() >= n {

                break;

            }

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

        }

        assert_eq!(runs.len(), n, "runs");

        let mut j: Vec<String> = db::flow::jobs(&app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();

        j.sort();

        j

    };

    assert_eq!(names(h.app.clone(), 1).await, ["build", "dist", "lint", "test"], "the first run has every job");

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

    assert_eq!(names(h.app.clone(), 2).await, ["build", "lint", "test"], "dist is out of scope");

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

    assert_eq!(names(h.app.clone(), 3).await, ["build", "dist", "lint", "test"], "a version bump runs dist");

}