crates/hub/src/db/flow.rs at d7ce41acf22c · andymccall/oxygenhub

# d7ce41acf22c oxygenhub / crates / hub / src / db / flow.rs

20.1 KiB · Rust · last changed Runs started by a push honour each job's scope 1 month ago source blame history raw

//! OxygenFlow runs, jobs and logs (hub spec §13.2).

use super::{now_ms, Pool};
use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]
pub struct Run {
    pub id: i64,
    pub repo_id: i64,
    /// The run's number within its repository, from 1 (migration 0019).
    #[sqlx(default)]
    pub number: i64,
    /// `push`, `manual` or `rerun:N` (migration 0020).
    #[sqlx(default)]
    pub trigger: String,
    pub bookmark: String,
    pub state: String,
    pub heads: String,
    pub pipeline_hash: String,
    pub status: String,
    pub error: String,
    pub actor: String,
    pub created_ms: i64,
    pub finished_ms: Option<i64>,
}

impl Run {
    /// How the run page says what started it.
    pub fn started_by(&self) -> String {
        match self.trigger.as_str() {
            "manual" => format!("started by {}", self.actor),
            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", &t[6..], self.actor),
            _ => format!("pushed by {}", self.actor),
        }
    }

    pub fn head_list(&self) -> Vec<&str> {
        self.heads.split('+').filter(|h| !h.is_empty()).collect()
    }

    pub fn live(&self) -> bool {
        matches!(self.status.as_str(), "queued" | "running")
    }
}

#[derive(Clone, Debug, FromRow)]
pub struct Job {
    pub id: i64,
    pub run_id: i64,
    pub name: String,
    pub depth: i64,
    pub depends: String,
    pub spec: String,
    pub sandbox: String,
    pub runtime: String,
    pub status: String,
    pub runner: String,
    pub token_hash: String,
    pub started_ms: Option<i64>,
    pub finished_ms: Option<i64>,
    pub exit_code: Option<i64>,
    pub error: String,
    pub receipt: String,
    pub receipt_hash: String,
    pub log_count: i64,
    /// Comma-separated ids of other repositories the job token may read.
    pub also_read: String,
    /// The receipt attestation kept in the repository (Oxygen spec §14.6).
    pub receipt_attest: String,
    /// Who signed it: the runner's identity name.
    pub receipt_signer: String,
    /// Why a signed receipt was not kept.
    pub receipt_note: String,
}

impl Job {
    pub fn depends_on(&self) -> Vec<String> {
        serde_json::from_str(&self.depends).unwrap_or_default()
    }

    pub fn done(&self) -> bool {
        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error")
    }
}

#[derive(Clone, Debug, FromRow)]
pub struct LogLine {
    pub seq: i64,
    pub t: i64,
    pub stream: String,
    pub text: String,
}

/// A job as the trigger creates it.
pub struct NewJob {
    pub name: String,
    pub depth: i64,
    pub depends: Vec<String>,
    pub spec: String,
    pub sandbox: String,
    pub runtime: String,
}

pub struct NewRun<'a> {
    pub repo_id: i64,
    pub bookmark: &'a str,
    pub state: &'a str,
    pub heads: &'a str,
    pub pipeline_hash: &'a str,
    pub actor: &'a str,
    /// `push`, `manual` or `rerun:N`.
    pub trigger: &'a str,
    /// Problems when the pipeline is invalid: the run is recorded in error.
    pub error: Option<&'a str>,
}

pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {
    let mut tx = super::begin_write(pool).await?;
    let now = now_ms();
    let status = if r.error.is_some() { "error" } else { "queued" };
    let id: i64 = sqlx::query_scalar(
        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
         VALUES (?, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
    )
    .bind(r.repo_id)
    .bind(r.repo_id)
    .bind(r.bookmark)
    .bind(r.state)
    .bind(r.heads)
    .bind(r.pipeline_hash)
    .bind(status)
    .bind(r.error.unwrap_or(""))
    .bind(r.actor)
    .bind(r.trigger)
    .bind(now)
    .bind(r.error.map(|_| now))
    .fetch_one(&mut *tx)
    .await?;
    for j in jobs {
        sqlx::query(
            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued')",
        )
        .bind(id)
        .bind(&j.name)
        .bind(j.depth)
        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))
        .bind(&j.spec)
        .bind(&j.sandbox)
        .bind(&j.runtime)
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(id)
}

/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).
fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {
    let mut out: Vec<String> = [spec, sandbox]
        .iter()
        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())
        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())
        .filter_map(|l| l.as_str().map(String::from))
        .collect();
    out.sort();
    out.dedup();
    out
}

impl Job {
    /// The runner labels this job needs.
    pub fn needs(&self) -> Vec<String> {
        labels_in(&self.spec, &self.sandbox)
    }
}

/// A job waiting for or held by a runner, with where it comes from.
#[derive(Clone, Debug, FromRow)]
pub struct QueueEntry {
    pub job_id: i64,
    pub run_id: i64,
    pub repo_id: i64,
    pub owner: String,
    pub name: String,
    pub visibility: String,
    pub run_number: i64,
    pub job: String,
    pub status: String,
    /// The runner holding it, when running.
    pub runner: String,
    pub spec: String,
    pub sandbox: String,
}

impl QueueEntry {
    /// The runner labels the job needs.
    pub fn needs(&self) -> Vec<String> {
        labels_in(&self.spec, &self.sandbox)
    }
}

/// Every job of a live run that is running or queued: running ones first,
/// then queued ones in the order runners claim them (hub spec §13.2).
pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
    sqlx::query_as(
        "SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox \
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
         WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \
         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, j.run_id, j.depth, j.id",
    )
    .fetch_all(pool)
    .await
}

/// A repository's runs, newest first, numbered below `before`.
pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number < ? ORDER BY number DESC LIMIT ?")
        .bind(repo_id)
        .bind(before.unwrap_or(i64::MAX))
        .bind(limit)
        .fetch_all(pool)
        .await
}

/// 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
}

pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND id < ? ORDER BY id DESC LIMIT ?")
        .bind(repo_id)
        .bind(before.unwrap_or(i64::MAX))
        .bind(limit)
        .fetch_all(pool)
        .await
}

/// A repository's run by its number there.
pub async fn run(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<Run>> {
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number = ?").bind(repo_id).bind(number).fetch_optional(pool).await
}

pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {
    sqlx::query_as("SELECT * FROM flow_runs WHERE id = ?").bind(id).fetch_optional(pool).await
}

pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {
    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ? ORDER BY depth, name").bind(run_id).fetch_all(pool).await
}

pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {
    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(id).fetch_optional(pool).await
}

/// The running job a job token belongs to, while it may still be used.
pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {
    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = ? AND status = 'running' AND started_ms >= ?")
        .bind(token_hash)
        .bind(now_ms() - max_age_ms)
        .fetch_optional(pool)
        .await
}

/// Take the oldest queued job whose dependencies all passed and whose tier
/// the runner offers, and mark it running for `runner` with a token.
/// Let a job's token also read these repositories (its pinned subrepos).
pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {
    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();
    sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;
    Ok(())
}

impl Job {
    pub fn also_read_ids(&self) -> Vec<i64> {
        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()
    }
}

/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no
/// limit, and the rootfs images it holds, `None` when it does not say.
#[derive(Clone, Debug, Default)]
pub struct Room {
    pub memory_mib: Option<u32>,
    pub cpus: Option<u32>,
    pub images: Option<std::collections::HashSet<String>>,
}

impl Room {
    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,
    /// and its rootfs image among the runner's (oxygenhub#73).
    pub fn fits(&self, sandbox_json: &str) -> bool {
        let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();
        let need = |k: &str| v[k].as_u64().unwrap_or(0);
        let rootfs = v["rootfs"].as_str().unwrap_or("");
        self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)
            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)
            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)))
    }
}

pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {
    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await
}

/// The oldest runnable job that fits in `room`.
/// The first queued job a runner with these tiers, this room and this
/// scope may take, now running under `runner`. A scoped runner never sees
/// another scope's jobs (oxygenhub#41).
pub async fn claim_within(
    pool: &Pool,
    runtimes: &[String],
    runner: &str,
    token_hash: &str,
    room: Room,
    scope: &super::runners::Scope,
    labels: &[String],
) -> sqlx::Result<Option<Job>> {
    let mut tx = super::begin_write(pool).await?;
    let queued: Vec<Job> = sqlx::query_as(
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \
         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') \
         AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \
         ORDER BY j.run_id, j.depth, j.id LIMIT 200",
    )
    .bind(&scope.owner)
    .bind(&scope.owner)
    .bind(scope.repo_id)
    .bind(scope.repo_id)
    .fetch_all(&mut *tx)
    .await?;
    for j in queued {
        if !runtimes.iter().any(|r| r == &j.runtime) || !room.fits(&j.sandbox) {
            continue;
        }
        // Every label the job and its sandbox ask for (`runs-on`).
        if !j.needs().iter().all(|l| labels.contains(l)) {
            continue;
        }
        let deps = j.depends_on();
        if !deps.is_empty() {
            let mut passed = 0i64;
            for d in &deps {
                let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")
                    .bind(j.run_id)
                    .bind(d)
                    .fetch_optional(&mut *tx)
                    .await?;
                if st.as_deref() == Some("passed") {
                    passed += 1;
                }
            }
            if passed != deps.len() as i64 {
                continue;
            }
        }
        let now = now_ms();
        let done = sqlx::query(
            "UPDATE flow_jobs SET status = 'running', runner = ?, token_hash = ?, started_ms = ? WHERE id = ? AND status = 'queued'",
        )
        .bind(runner)
        .bind(token_hash)
        .bind(now)
        .bind(j.id)
        .execute(&mut *tx)
        .await?;
        if done.rows_affected() == 1 {
            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = ? AND status = 'queued'")
                .bind(j.run_id)
                .execute(&mut *tx)
                .await?;
            tx.commit().await?;
            return job(pool, j.id).await;
        }
    }
    tx.commit().await?;
    Ok(None)
}

/// Append records; returns how many were kept (the rest exceed the cap).
pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {
    let mut tx = super::begin_write(pool).await?;
    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
    let room = (cap - count).max(0) as usize;
    let keep = &records[..records.len().min(room)];
    for (i, r) in keep.iter().enumerate() {
        sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES (?, ?, ?, ?, ?)")
            .bind(job_id)
            .bind(count + i as i64 + 1)
            .bind(r.t as i64)
            .bind(&r.stream)
            .bind(&r.text)
            .execute(&mut *tx)
            .await?;
    }
    sqlx::query("UPDATE flow_jobs SET log_count = log_count + ? WHERE id = ?")
        .bind(keep.len() as i64)
        .bind(job_id)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(keep.len())
}

pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {
    sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")
        .bind(job_id)
        .bind(after)
        .bind(limit)
        .fetch_all(pool)
        .await
}

/// Jobs running since before `ms` (the oldest first).
pub async fn running_since_before(pool: &Pool, ms: i64) -> sqlx::Result<Vec<Job>> {
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < ? ORDER BY started_ms")
        .bind(ms)
        .fetch_all(pool)
        .await
}

/// The jobs running on the runner with this id.
pub async fn running_on(pool: &Pool, runner_id: i64) -> sqlx::Result<Vec<Job>> {
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = ? ORDER BY id").bind(runner_id).fetch_all(pool).await
}

pub struct Outcome<'a> {
    pub status: &'a str,
    pub exit_code: Option<i64>,
    pub error: &'a str,
    pub receipt: &'a str,
    pub receipt_hash: &'a str,
}

/// Record a job's end, skip the jobs that can no longer run, and settle the
/// run. Returns the jobs whose status changed (for their checks).
pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {
    let mut tx = super::begin_write(pool).await?;
    let now = now_ms();
    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
    sqlx::query("UPDATE flow_jobs SET status = ?, exit_code = ?, error = ?, receipt = ?, receipt_hash = ?, finished_ms = ?, token_hash = '' WHERE id = ?")
        .bind(o.status)
        .bind(o.exit_code)
        .bind(o.error)
        .bind(o.receipt)
        .bind(o.receipt_hash)
        .bind(now)
        .bind(job_id)
        .execute(&mut *tx)
        .await?;
    let mut changed = vec![job_id];
    // Dependents of anything that did not pass are skipped, transitively.
    loop {
        let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
        let blocked: Vec<i64> = all
            .iter()
            .filter(|x| x.status == "queued")
            .filter(|x| {
                x.depends_on()
                    .iter()
                    .any(|d| all.iter().any(|y| &y.name == d && matches!(y.status.as_str(), "failed" | "skipped" | "error")))
            })
            .map(|x| x.id)
            .collect();
        if blocked.is_empty() {
            break;
        }
        for id in blocked {
            sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = ? WHERE id = ?")
                .bind(now)
                .bind(id)
                .execute(&mut *tx)
                .await?;
            changed.push(id);
        }
    }
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
    if all.iter().all(Job::done) {
        let status = if all.iter().all(|x| x.status == "passed") { "passed" } else { "failed" };
        sqlx::query("UPDATE flow_runs SET status = ?, finished_ms = ? WHERE id = ?")
            .bind(status)
            .bind(now)
            .bind(j.run_id)
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    let mut out = Vec::new();
    for id in changed {
        if let Some(x) = job(pool, id).await? {
            out.push(x);
        }
    }
    Ok(out)
}

/// The newest run of exactly this state (heads joined with `+`, sorted).
pub async fn latest_for_heads(pool: &Pool, repo_id: i64, heads: &str) -> sqlx::Result<Option<Run>> {
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND heads = ? ORDER BY id DESC LIMIT 1")
        .bind(repo_id)
        .bind(heads)
        .fetch_optional(pool)
        .await
}

/// Record what became of job `job_id`'s signed receipt.
pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_jobs SET receipt_attest = ?, receipt_signer = ?, receipt_note = ? WHERE id = ?")
        .bind(attest)
        .bind(signer)
        .bind(note)
        .bind(job_id)
        .execute(pool)
        .await?;
    Ok(())
}

/// A job's artifact, kept as a blob of the run's repository.
#[derive(Clone, Debug, FromRow)]
pub struct Artifact {
    pub job_id: i64,
    pub name: String,
    pub path: String,
    pub blob: String,
    pub size: i64,
    /// BLAKE3 of the bytes.
    pub content: String,
    pub created_ms: i64,
}

pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {
    sqlx::query("INSERT OR REPLACE INTO flow_artifacts (job_id, name, path, blob, size, content, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?)")
        .bind(a.job_id)
        .bind(&a.name)
        .bind(&a.path)
        .bind(&a.blob)
        .bind(a.size)
        .bind(&a.content)
        .bind(a.created_ms)
        .execute(pool)
        .await?;
    Ok(())
}

/// Job `job_id`'s artifacts, by name.
pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = ? ORDER BY name").bind(job_id).fetch_all(pool).await
}

/// Where an artifact with given content came from: the run and job that
/// built it, and the job's signed receipt when one was kept.
#[derive(Clone, Debug, FromRow)]
pub struct Provenance {
    pub run_number: i64,
    pub job: String,
    pub name: String,
    pub receipt_attest: String,
    pub receipt_signer: String,
}

/// The build of the bytes whose BLAKE3 is `content` in repository
/// `repo_id`: a passed job, one with a kept signed receipt first, newest
/// run first.
pub async fn provenance(pool: &Pool, repo_id: i64, content: &str) -> sqlx::Result<Option<Provenance>> {
    sqlx::query_as(
        "SELECT r.number AS run_number, j.name AS job, a.name AS name, j.receipt_attest, j.receipt_signer \
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id \
         WHERE r.repo_id = ? AND a.content = ? AND j.status = 'passed' \
         ORDER BY (j.receipt_attest != '') DESC, r.number DESC LIMIT 1",
    )
    .bind(repo_id)
    .bind(content)
    .fetch_optional(pool)
    .await
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1	//! OxygenFlow runs, jobs and logs (hub spec §13.2).
2
3	use super::{now_ms, Pool};
4	use sqlx::FromRow;
5
6	#[derive(Clone, Debug, FromRow)]
7	pub struct Run {
8	pub id: i64,
9	pub repo_id: i64,
10	/// The run's number within its repository, from 1 (migration 0019).
11	#[sqlx(default)]
12	pub number: i64,
13	/// `push`, `manual` or `rerun:N` (migration 0020).
14	#[sqlx(default)]
15	pub trigger: String,
16	pub bookmark: String,
17	pub state: String,
18	pub heads: String,
19	pub pipeline_hash: String,
20	pub status: String,
21	pub error: String,
22	pub actor: String,
23	pub created_ms: i64,
24	pub finished_ms: Option<i64>,
25	}
26
27	impl Run {
28	/// How the run page says what started it.
29	pub fn started_by(&self) -> String {
30	match self.trigger.as_str() {
31	"manual" => format!("started by {}", self.actor),
32	t if t.starts_with("rerun:") => format!("re-run of #{} by {}", &t[6..], self.actor),
33	_ => format!("pushed by {}", self.actor),
34	}
35	}
36
37	pub fn head_list(&self) -> Vec<&str> {
38	self.heads.split('+').filter(\|h\| !h.is_empty()).collect()
39	}
40
41	pub fn live(&self) -> bool {
42	matches!(self.status.as_str(), "queued" \| "running")
43	}
44	}
45
46	#[derive(Clone, Debug, FromRow)]
47	pub struct Job {
48	pub id: i64,
49	pub run_id: i64,
50	pub name: String,
51	pub depth: i64,
52	pub depends: String,
53	pub spec: String,
54	pub sandbox: String,
55	pub runtime: String,
56	pub status: String,
57	pub runner: String,
58	pub token_hash: String,
59	pub started_ms: Option<i64>,
60	pub finished_ms: Option<i64>,
61	pub exit_code: Option<i64>,
62	pub error: String,
63	pub receipt: String,
64	pub receipt_hash: String,
65	pub log_count: i64,
66	/// Comma-separated ids of other repositories the job token may read.
67	pub also_read: String,
68	/// The receipt attestation kept in the repository (Oxygen spec §14.6).
69	pub receipt_attest: String,
70	/// Who signed it: the runner's identity name.
71	pub receipt_signer: String,
72	/// Why a signed receipt was not kept.
73	pub receipt_note: String,
74	}
75
76	impl Job {
77	pub fn depends_on(&self) -> Vec<String> {
78	serde_json::from_str(&self.depends).unwrap_or_default()
79	}
80
81	pub fn done(&self) -> bool {
82	matches!(self.status.as_str(), "passed" \| "failed" \| "skipped" \| "error")
83	}
84	}
85
86	#[derive(Clone, Debug, FromRow)]
87	pub struct LogLine {
88	pub seq: i64,
89	pub t: i64,
90	pub stream: String,
91	pub text: String,
92	}
93
94	/// A job as the trigger creates it.
95	pub struct NewJob {
96	pub name: String,
97	pub depth: i64,
98	pub depends: Vec<String>,
99	pub spec: String,
100	pub sandbox: String,
101	pub runtime: String,
102	}
103
104	pub struct NewRun<'a> {
105	pub repo_id: i64,
106	pub bookmark: &'a str,
107	pub state: &'a str,
108	pub heads: &'a str,
109	pub pipeline_hash: &'a str,
110	pub actor: &'a str,
111	/// `push`, `manual` or `rerun:N`.
112	pub trigger: &'a str,
113	/// Problems when the pipeline is invalid: the run is recorded in error.
114	pub error: Option<&'a str>,
115	}
116
117	pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {
118	let mut tx = super::begin_write(pool).await?;
119	let now = now_ms();
120	let status = if r.error.is_some() { "error" } else { "queued" };
121	let id: i64 = sqlx::query_scalar(
122	"INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
123	VALUES (?, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
124	)
125	.bind(r.repo_id)
126	.bind(r.repo_id)
127	.bind(r.bookmark)
128	.bind(r.state)
129	.bind(r.heads)
130	.bind(r.pipeline_hash)
131	.bind(status)
132	.bind(r.error.unwrap_or(""))
133	.bind(r.actor)
134	.bind(r.trigger)
135	.bind(now)
136	.bind(r.error.map(\|_\| now))
137	.fetch_one(&mut *tx)
138	.await?;
139	for j in jobs {
140	sqlx::query(
141	"INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued')",
142	)
143	.bind(id)
144	.bind(&j.name)
145	.bind(j.depth)
146	.bind(serde_json::to_string(&j.depends).unwrap_or_else(\|_\| "[]".into()))
147	.bind(&j.spec)
148	.bind(&j.sandbox)
149	.bind(&j.runtime)
150	.execute(&mut *tx)
151	.await?;
152	}
153	tx.commit().await?;
154	Ok(id)
155	}
156
157	/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).
158	fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {
159	let mut out: Vec<String> = [spec, sandbox]
160	.iter()
161	.filter_map(\|j\| serde_json::from_str::<serde_json::Value>(j).ok())
162	.flat_map(\|v\| v["runs_on"].as_array().cloned().unwrap_or_default())
163	.filter_map(\|l\| l.as_str().map(String::from))
164	.collect();
165	out.sort();
166	out.dedup();
167	out
168	}
169
170	impl Job {
171	/// The runner labels this job needs.
172	pub fn needs(&self) -> Vec<String> {
173	labels_in(&self.spec, &self.sandbox)
174	}
175	}
176
177	/// A job waiting for or held by a runner, with where it comes from.
178	#[derive(Clone, Debug, FromRow)]
179	pub struct QueueEntry {
180	pub job_id: i64,
181	pub run_id: i64,
182	pub repo_id: i64,
183	pub owner: String,
184	pub name: String,
185	pub visibility: String,
186	pub run_number: i64,
187	pub job: String,
188	pub status: String,
189	/// The runner holding it, when running.
190	pub runner: String,
191	pub spec: String,
192	pub sandbox: String,
193	}
194
195	impl QueueEntry {
196	/// The runner labels the job needs.
197	pub fn needs(&self) -> Vec<String> {
198	labels_in(&self.spec, &self.sandbox)
199	}
200	}
201
202	/// Every job of a live run that is running or queued: running ones first,
203	/// then queued ones in the order runners claim them (hub spec §13.2).
204	pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
205	sqlx::query_as(
206	"SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox \
207	FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
208	WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \
209	ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, j.run_id, j.depth, j.id",
210	)
211	.fetch_all(pool)
212	.await
213	}
214
215	/// A repository's runs, newest first, numbered below `before`.
216	pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
217	sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number < ? ORDER BY number DESC LIMIT ?")
218	.bind(repo_id)
219	.bind(before.unwrap_or(i64::MAX))
220	.bind(limit)
221	.fetch_all(pool)
222	.await
223	}
224
225	/// The latest run on this bookmark of this repository, for what a push
226	/// changed since (a job's scope).
227	pub async fn latest_on(pool: &Pool, repo_id: i64, bookmark: &str) -> sqlx::Result<Option<Run>> {
228	sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND heads != '' ORDER BY id DESC LIMIT 1")
229	.bind(repo_id)
230	.bind(bookmark)
231	.fetch_optional(pool)
232	.await
233	}
234
235	pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
236	sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND id < ? ORDER BY id DESC LIMIT ?")
237	.bind(repo_id)
238	.bind(before.unwrap_or(i64::MAX))
239	.bind(limit)
240	.fetch_all(pool)
241	.await
242	}
243
244	/// A repository's run by its number there.
245	pub async fn run(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<Run>> {
246	sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number = ?").bind(repo_id).bind(number).fetch_optional(pool).await
247	}
248
249	pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {
250	sqlx::query_as("SELECT * FROM flow_runs WHERE id = ?").bind(id).fetch_optional(pool).await
251	}
252
253	pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {
254	sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ? ORDER BY depth, name").bind(run_id).fetch_all(pool).await
255	}
256
257	pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {
258	sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(id).fetch_optional(pool).await
259	}
260
261	/// The running job a job token belongs to, while it may still be used.
262	pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {
263	sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = ? AND status = 'running' AND started_ms >= ?")
264	.bind(token_hash)
265	.bind(now_ms() - max_age_ms)
266	.fetch_optional(pool)
267	.await
268	}
269
270	/// Take the oldest queued job whose dependencies all passed and whose tier
271	/// the runner offers, and mark it running for `runner` with a token.
272	/// Let a job's token also read these repositories (its pinned subrepos).
273	pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {
274	let list: Vec<String> = repo_ids.iter().map(\|i\| i.to_string()).collect();
275	sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;
276	Ok(())
277	}
278
279	impl Job {
280	pub fn also_read_ids(&self) -> Vec<i64> {
281	self.also_read.split(',').filter_map(\|x\| x.trim().parse().ok()).collect()
282	}
283	}
284
285	/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no
286	/// limit, and the rootfs images it holds, `None` when it does not say.
287	#[derive(Clone, Debug, Default)]
288	pub struct Room {
289	pub memory_mib: Option<u32>,
290	pub cpus: Option<u32>,
291	pub images: Option<std::collections::HashSet<String>>,
292	}
293
294	impl Room {
295	/// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,
296	/// and its rootfs image among the runner's (oxygenhub#73).
297	pub fn fits(&self, sandbox_json: &str) -> bool {
298	let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();
299	let need = \|k: &str\| v[k].as_u64().unwrap_or(0);
300	let rootfs = v["rootfs"].as_str().unwrap_or("");
301	self.memory_mib.is_none_or(\|m\| need("memory_mib") <= m as u64)
302	&& self.cpus.is_none_or(\|c\| need("cpus") <= c as u64)
303	&& (rootfs.is_empty() \|\| self.images.as_ref().is_none_or(\|have\| have.contains(rootfs)))
304	}
305	}
306
307	pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {
308	claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await
309	}
310
311	/// The oldest runnable job that fits in `room`.
312	/// The first queued job a runner with these tiers, this room and this
313	/// scope may take, now running under `runner`. A scoped runner never sees
314	/// another scope's jobs (oxygenhub#41).
315	pub async fn claim_within(
316	pool: &Pool,
317	runtimes: &[String],
318	runner: &str,
319	token_hash: &str,
320	room: Room,
321	scope: &super::runners::Scope,
322	labels: &[String],
323	) -> sqlx::Result<Option<Job>> {
324	let mut tx = super::begin_write(pool).await?;
325	let queued: Vec<Job> = sqlx::query_as(
326	"SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \
327	WHERE j.status = 'queued' AND r.status IN ('queued', 'running') \
328	AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \
329	ORDER BY j.run_id, j.depth, j.id LIMIT 200",
330	)
331	.bind(&scope.owner)
332	.bind(&scope.owner)
333	.bind(scope.repo_id)
334	.bind(scope.repo_id)
335	.fetch_all(&mut *tx)
336	.await?;
337	for j in queued {
338	if !runtimes.iter().any(\|r\| r == &j.runtime) \|\| !room.fits(&j.sandbox) {
339	continue;
340	}
341	// Every label the job and its sandbox ask for (`runs-on`).
342	if !j.needs().iter().all(\|l\| labels.contains(l)) {
343	continue;
344	}
345	let deps = j.depends_on();
346	if !deps.is_empty() {
347	let mut passed = 0i64;
348	for d in &deps {
349	let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")
350	.bind(j.run_id)
351	.bind(d)
352	.fetch_optional(&mut *tx)
353	.await?;
354	if st.as_deref() == Some("passed") {
355	passed += 1;
356	}
357	}
358	if passed != deps.len() as i64 {
359	continue;
360	}
361	}
362	let now = now_ms();
363	let done = sqlx::query(
364	"UPDATE flow_jobs SET status = 'running', runner = ?, token_hash = ?, started_ms = ? WHERE id = ? AND status = 'queued'",
365	)
366	.bind(runner)
367	.bind(token_hash)
368	.bind(now)
369	.bind(j.id)
370	.execute(&mut *tx)
371	.await?;
372	if done.rows_affected() == 1 {
373	sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = ? AND status = 'queued'")
374	.bind(j.run_id)
375	.execute(&mut *tx)
376	.await?;
377	tx.commit().await?;
378	return job(pool, j.id).await;
379	}
380	}
381	tx.commit().await?;
382	Ok(None)
383	}
384
385	/// Append records; returns how many were kept (the rest exceed the cap).
386	pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {
387	let mut tx = super::begin_write(pool).await?;
388	let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
389	let room = (cap - count).max(0) as usize;
390	let keep = &records[..records.len().min(room)];
391	for (i, r) in keep.iter().enumerate() {
392	sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES (?, ?, ?, ?, ?)")
393	.bind(job_id)
394	.bind(count + i as i64 + 1)
395	.bind(r.t as i64)
396	.bind(&r.stream)
397	.bind(&r.text)
398	.execute(&mut *tx)
399	.await?;
400	}
401	sqlx::query("UPDATE flow_jobs SET log_count = log_count + ? WHERE id = ?")
402	.bind(keep.len() as i64)
403	.bind(job_id)
404	.execute(&mut *tx)
405	.await?;
406	tx.commit().await?;
407	Ok(keep.len())
408	}
409
410	pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {
411	sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")
412	.bind(job_id)
413	.bind(after)
414	.bind(limit)
415	.fetch_all(pool)
416	.await
417	}
418
419	/// Jobs running since before `ms` (the oldest first).
420	pub async fn running_since_before(pool: &Pool, ms: i64) -> sqlx::Result<Vec<Job>> {
421	sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < ? ORDER BY started_ms")
422	.bind(ms)
423	.fetch_all(pool)
424	.await
425	}
426
427	/// The jobs running on the runner with this id.
428	pub async fn running_on(pool: &Pool, runner_id: i64) -> sqlx::Result<Vec<Job>> {
429	sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = ? ORDER BY id").bind(runner_id).fetch_all(pool).await
430	}
431
432	pub struct Outcome<'a> {
433	pub status: &'a str,
434	pub exit_code: Option<i64>,
435	pub error: &'a str,
436	pub receipt: &'a str,
437	pub receipt_hash: &'a str,
438	}
439
440	/// Record a job's end, skip the jobs that can no longer run, and settle the
441	/// run. Returns the jobs whose status changed (for their checks).
442	pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {
443	let mut tx = super::begin_write(pool).await?;
444	let now = now_ms();
445	let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
446	sqlx::query("UPDATE flow_jobs SET status = ?, exit_code = ?, error = ?, receipt = ?, receipt_hash = ?, finished_ms = ?, token_hash = '' WHERE id = ?")
447	.bind(o.status)
448	.bind(o.exit_code)
449	.bind(o.error)
450	.bind(o.receipt)
451	.bind(o.receipt_hash)
452	.bind(now)
453	.bind(job_id)
454	.execute(&mut *tx)
455	.await?;
456	let mut changed = vec![job_id];
457	// Dependents of anything that did not pass are skipped, transitively.
458	loop {
459	let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
460	let blocked: Vec<i64> = all
461	.iter()
462	.filter(\|x\| x.status == "queued")
463	.filter(\|x\| {
464	x.depends_on()
465	.iter()
466	.any(\|d\| all.iter().any(\|y\| &y.name == d && matches!(y.status.as_str(), "failed" \| "skipped" \| "error")))
467	})
468	.map(\|x\| x.id)
469	.collect();
470	if blocked.is_empty() {
471	break;
472	}
473	for id in blocked {
474	sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = ? WHERE id = ?")
475	.bind(now)
476	.bind(id)
477	.execute(&mut *tx)
478	.await?;
479	changed.push(id);
480	}
481	}
482	let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
483	if all.iter().all(Job::done) {
484	let status = if all.iter().all(\|x\| x.status == "passed") { "passed" } else { "failed" };
485	sqlx::query("UPDATE flow_runs SET status = ?, finished_ms = ? WHERE id = ?")
486	.bind(status)
487	.bind(now)
488	.bind(j.run_id)
489	.execute(&mut *tx)
490	.await?;
491	}
492	tx.commit().await?;
493	let mut out = Vec::new();
494	for id in changed {
495	if let Some(x) = job(pool, id).await? {
496	out.push(x);
497	}
498	}
499	Ok(out)
500	}
501
502	/// The newest run of exactly this state (heads joined with `+`, sorted).
503	pub async fn latest_for_heads(pool: &Pool, repo_id: i64, heads: &str) -> sqlx::Result<Option<Run>> {
504	sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND heads = ? ORDER BY id DESC LIMIT 1")
505	.bind(repo_id)
506	.bind(heads)
507	.fetch_optional(pool)
508	.await
509	}
510
511	/// Record what became of job `job_id`'s signed receipt.
512	pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {
513	sqlx::query("UPDATE flow_jobs SET receipt_attest = ?, receipt_signer = ?, receipt_note = ? WHERE id = ?")
514	.bind(attest)
515	.bind(signer)
516	.bind(note)
517	.bind(job_id)
518	.execute(pool)
519	.await?;
520	Ok(())
521	}
522
523	/// A job's artifact, kept as a blob of the run's repository.
524	#[derive(Clone, Debug, FromRow)]
525	pub struct Artifact {
526	pub job_id: i64,
527	pub name: String,
528	pub path: String,
529	pub blob: String,
530	pub size: i64,
531	/// BLAKE3 of the bytes.
532	pub content: String,
533	pub created_ms: i64,
534	}
535
536	pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {
537	sqlx::query("INSERT OR REPLACE INTO flow_artifacts (job_id, name, path, blob, size, content, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?)")
538	.bind(a.job_id)
539	.bind(&a.name)
540	.bind(&a.path)
541	.bind(&a.blob)
542	.bind(a.size)
543	.bind(&a.content)
544	.bind(a.created_ms)
545	.execute(pool)
546	.await?;
547	Ok(())
548	}
549
550	/// Job `job_id`'s artifacts, by name.
551	pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
552	sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = ? ORDER BY name").bind(job_id).fetch_all(pool).await
553	}
554
555	/// Where an artifact with given content came from: the run and job that
556	/// built it, and the job's signed receipt when one was kept.
557	#[derive(Clone, Debug, FromRow)]
558	pub struct Provenance {
559	pub run_number: i64,
560	pub job: String,
561	pub name: String,
562	pub receipt_attest: String,
563	pub receipt_signer: String,
564	}
565
566	/// The build of the bytes whose BLAKE3 is `content` in repository
567	/// `repo_id`: a passed job, one with a kept signed receipt first, newest
568	/// run first.
569	pub async fn provenance(pool: &Pool, repo_id: i64, content: &str) -> sqlx::Result<Option<Provenance>> {
570	sqlx::query_as(
571	"SELECT r.number AS run_number, j.name AS job, a.name AS name, j.receipt_attest, j.receipt_signer \
572	FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id \
573	WHERE r.repo_id = ? AND a.content = ? AND j.status = 'passed' \
574	ORDER BY (j.receipt_attest != '') DESC, r.number DESC LIMIT 1",
575	)
576	.bind(repo_id)
577	.bind(content)
578	.fetch_optional(pool)
579	.await
580	}

Keyboard