OXYGENHUB
◆ oh-comment-body-file oxygenhub / crates / hub / src / db / flow.rs
48.7 KiB · Rust · last changed Insights: OxygenFlow's standard charts, for the site, an owner and a repository 17 days ago source blame history raw
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 
    /// `urgent` when expedited (oxygenhub#244), else empty: the run follows
26 
    /// its repository's, organisation's or the site's level.
27 
    #[sqlx(default)]
28 
    pub priority: String,
29 
    /// Who expedited it and why.
30 
    #[sqlx(default)]
31 
    pub priority_note: String,
32 
}
33
34 
/// The priority levels, most urgent first (oxygenhub#244).
35 
pub const PRIORITIES: [&str; 4] = ["urgent", "high", "normal", "low"];
36
37 
/// SQL for the level a run's jobs get from where they are, as a rank
38 
/// (0 urgent … 3 low): the run's own (expedited), else its repository's,
39 
/// its organisation's, the site's (`settings.flow_priority`), else normal.
40 
macro_rules! level_rank {
41 
    () => {
42 
        "CASE COALESCE(NULLIF(r.priority, ''), NULLIF(rep.flow_priority, ''), \
43 
           NULLIF((SELECT o.flow_priority FROM organisations o WHERE o.name = rep.owner), ''), \
44 
           NULLIF((SELECT s.value FROM settings s WHERE s.key = 'flow_priority'), ''), 'normal') \
45 
         WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'low' THEN 3 ELSE 2 END"
46 
    };
47 
}
48
49 
/// SQL for a queued job's level as a rank (0 urgent … 3 low): the level
50 
/// its run gets from where it is, lowered by the job's own priority from
51 
/// its pipeline (`low`, `normal`), never raised by it; an expedited run is
52 
/// urgent whatever its jobs say (oxygenhub#244). Needs `j` (the job), `r`
53 
/// (its run) and `rep` (its repository) in the query.
54 
pub const PRIORITY_RANK: &str = concat!(
55 
    "CASE WHEN r.priority <> '' THEN ",
56 
    level_rank!(),
57 
    " WHEN j.priority = 'low' THEN 3 WHEN j.priority = 'normal' AND (",
58 
    level_rank!(),
59 
    ") < 2 THEN 2 ELSE ",
60 
    level_rank!(),
61 
    " END"
62 
);
63
64 
/// SQL for why a job stands where it does, beside [`PRIORITY_RANK`]: an
65 
/// expedited run's note, else the pipeline's word for a job it lowered.
66 
pub const PRIORITY_NOTE: &str =
67 
    "CASE WHEN r.priority <> '' THEN r.priority_note WHEN j.priority <> '' THEN 'pipeline: ' || j.priority ELSE r.priority_note END";
68
69 
/// How many runs of `owner`'s repositories are expedited and still live.
70 
pub async fn expedited_live(pool: &Pool, owner: &str) -> sqlx::Result<i64> {
71 
    sqlx::query_scalar(
72 
        "SELECT COUNT(*) FROM flow_runs r JOIN repositories rep ON rep.id = r.repo_id \
73 
         WHERE rep.owner = $1 AND r.priority = 'urgent' AND r.status IN ('queued', 'running')",
74 
    )
75 
    .bind(owner)
76 
    .fetch_one(pool)
77 
    .await
78 
}
79
80 
/// Raise run `run_id` to urgent, saying who and why; a run already ended
81 
/// is left alone. Whether it changed.
82 
pub async fn expedite(pool: &Pool, run_id: i64, note: &str) -> sqlx::Result<bool> {
83 
    let r = sqlx::query("UPDATE flow_runs SET priority = 'urgent', priority_note = $1 WHERE id = $2 AND status IN ('queued', 'running')")
84 
        .bind(note)
85 
        .bind(run_id)
86 
        .execute(pool)
87 
        .await?;
88 
    Ok(r.rows_affected() > 0)
89 
}
90
91 
impl Run {
92 
    /// How the run page says what started it.
93 
    pub fn started_by(&self) -> String {
94 
        match self.trigger.as_str() {
95 
            "manual" => format!("started by {}", self.actor),
96 
            "schedule" => "started on schedule".to_string(),
97 
            t if t.starts_with("tag:") => format!("tag {} pushed by {}", &t[4..], self.actor),
98 
            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", self.rerun_of().unwrap_or(""), self.actor),
99 
            _ => format!("pushed by {}", self.actor),
100 
        }
101 
    }
102
103 
    /// The run a rerun repeats: `N` of `rerun:N` or `rerun:N:EVENT`.
104 
    pub fn rerun_of(&self) -> Option<&str> {
105 
        let rest = self.trigger.strip_prefix("rerun:")?;
106 
        Some(rest.split(':').next().unwrap_or(rest))
107 
    }
108
109 
    /// The trigger as people read it: `rerun:N`, without the event it repeats.
110 
    pub fn trigger_label(&self) -> String {
111 
        match self.rerun_of() {
112 
            Some(n) => format!("rerun:{n}"),
113 
            None => self.trigger.clone(),
114 
        }
115 
    }
116
117 
    pub fn head_list(&self) -> Vec<&str> {
118 
        self.heads.split('+').filter(|h| !h.is_empty()).collect()
119 
    }
120
121 
    pub fn live(&self) -> bool {
122 
        matches!(self.status.as_str(), "queued" | "running")
123 
    }
124 
}
125
126 
#[derive(Clone, Debug, Default, FromRow)]
127 
pub struct Job {
128 
    pub id: i64,
129 
    pub run_id: i64,
130 
    pub name: String,
131 
    pub depth: i64,
132 
    pub depends: String,
133 
    pub spec: String,
134 
    pub sandbox: String,
135 
    pub runtime: String,
136 
    pub status: String,
137 
    pub runner: String,
138 
    pub token_hash: String,
139 
    pub started_ms: Option<i64>,
140 
    pub finished_ms: Option<i64>,
141 
    pub exit_code: Option<i64>,
142 
    pub error: String,
143 
    pub receipt: String,
144 
    pub receipt_hash: String,
145 
    pub log_count: i64,
146 
    /// Comma-separated ids of other repositories the job token may read.
147 
    pub also_read: String,
148 
    /// The receipt attestation kept in the repository (Oxygen spec §14.6).
149 
    pub receipt_attest: String,
150 
    /// Who signed it: the runner's identity name.
151 
    pub receipt_signer: String,
152 
    /// Why a signed receipt was not kept.
153 
    pub receipt_note: String,
154 
    /// Why a queued job no runner can take waits (spec §11.4), from the
155 
    /// last placement pass; empty when it is only waiting its turn.
156 
    pub waiting: String,
157 
    /// What it took, as its runner reported (oxygenflow#41), in MiB.
158 
    pub peak_disk_mib: Option<i64>,
159 
    pub peak_memory_mib: Option<i64>,
160 
    /// The free disk it needs, learned from past runs of the same job.
161 
    pub need_disk_mib: Option<i64>,
162 
    /// Why it ran on its runner (oxygenflow#42).
163 
    pub placed: String,
164 
    /// `waiting` for a person, `approved`, or empty (oxygenflow#57).
165 
    pub approval: String,
166 
    /// Who approved or rejected it, and when.
167 
    pub approved_by: String,
168 
    pub approved_ms: Option<i64>,
169 
    /// The agent token it was decided through, if one (oxygenhub#229).
170 
    #[sqlx(default)]
171 
    pub approved_via: String,
172 
    /// The environment it deploys to, or empty (oxygenflow#59).
173 
    pub environment: String,
174 
    /// The approval attestation that released it, when sealed (#235).
175 
    pub approval_seal: String,
176 
    /// When it entered the queue (made, or run again), and when it became
177 
    /// ready to run: set at its claim (migration 0080, oxygenhub#257).
178 
    #[sqlx(default)]
179 
    pub queued_ms: Option<i64>,
180 
    #[sqlx(default)]
181 
    pub ready_ms: Option<i64>,
182 
}
183
184 
impl Job {
185 
    pub fn depends_on(&self) -> Vec<String> {
186 
        serde_json::from_str(&self.depends).unwrap_or_default()
187 
    }
188
189 
    pub fn done(&self) -> bool {
190 
        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error" | "cancelled")
191 
    }
192 
}
193
194 
#[derive(Clone, Debug, FromRow)]
195 
pub struct LogLine {
196 
    pub seq: i64,
197 
    pub t: i64,
198 
    pub stream: String,
199 
    pub text: String,
200 
}
201
202 
/// A job as the trigger creates it.
203 
pub struct NewJob {
204 
    pub name: String,
205 
    pub depth: i64,
206 
    pub depends: Vec<String>,
207 
    pub spec: String,
208 
    pub sandbox: String,
209 
    pub runtime: String,
210 
    /// The free disk it needs, from [`learned_disk_need`].
211 
    pub need_disk_mib: Option<i64>,
212 
}
213
214 
/// Runs of a job whose peaks set its next need.
215 
const NEED_FROM_RUNS: i64 = 5;
216
217 
/// The free disk a new run of job `name` in repository `repo_id` needs
218 
/// (oxygenflow#41): the most its last runs took, plus a quarter; `None`
219 
/// for a job that has not run here yet.
220 
pub async fn learned_disk_need(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<i64>> {
221 
    let peak: Option<i64> = sqlx::query_scalar(
222 
        "SELECT MAX(peak_disk_mib) FROM (SELECT j.peak_disk_mib FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
223 
         WHERE r.repo_id = $1 AND j.name = $2 AND j.peak_disk_mib IS NOT NULL ORDER BY j.id DESC LIMIT $3) recent",
224 
    )
225 
    .bind(repo_id)
226 
    .bind(name)
227 
    .bind(NEED_FROM_RUNS)
228 
    .fetch_one(pool)
229 
    .await?;
230 
    Ok(peak.map(|p| p + p / 4 + 64))
231 
}
232
233 
/// Record why job `job_id` ran where it did.
234 
pub async fn set_placed(pool: &Pool, job_id: i64, placed: &str) -> sqlx::Result<()> {
235 
    sqlx::query("UPDATE flow_jobs SET placed = $1 WHERE id = $2").bind(placed).bind(job_id).execute(pool).await?;
236 
    Ok(())
237 
}
238
239 
/// Keep what a job took, as its runner reported.
240 
pub async fn set_peaks(pool: &Pool, job_id: i64, disk_mib: Option<u64>, memory_mib: Option<u64>) -> sqlx::Result<()> {
241 
    sqlx::query("UPDATE flow_jobs SET peak_disk_mib = $1, peak_memory_mib = $2 WHERE id = $3")
242 
        .bind(disk_mib.map(|v| v.min(i64::MAX as u64) as i64))
243 
        .bind(memory_mib.map(|v| v.min(i64::MAX as u64) as i64))
244 
        .bind(job_id)
245 
        .execute(pool)
246 
        .await?;
247 
    Ok(())
248 
}
249
250 
pub struct NewRun<'a> {
251 
    pub repo_id: i64,
252 
    pub bookmark: &'a str,
253 
    pub state: &'a str,
254 
    pub heads: &'a str,
255 
    pub pipeline_hash: &'a str,
256 
    pub actor: &'a str,
257 
    /// `push`, `manual` or `rerun:N`.
258 
    pub trigger: &'a str,
259 
    /// Problems when the pipeline is invalid: the run is recorded in error.
260 
    pub error: Option<&'a str>,
261 
}
262
263 
pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {
264 
    let mut tx = super::begin_write(pool).await?;
265 
    let now = now_ms();
266 
    let status = if r.error.is_some() { "error" } else { "queued" };
267 
    let id: i64 = sqlx::query_scalar(
268 
        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
269 
         VALUES ($1, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = $2), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING id",
270 
    )
271 
    .bind(r.repo_id)
272 
    .bind(r.repo_id)
273 
    .bind(r.bookmark)
274 
    .bind(r.state)
275 
    .bind(r.heads)
276 
    .bind(r.pipeline_hash)
277 
    .bind(status)
278 
    .bind(r.error.unwrap_or(""))
279 
    .bind(r.actor)
280 
    .bind(r.trigger)
281 
    .bind(now)
282 
    .bind(r.error.map(|_| now))
283 
    .fetch_one(&mut *tx)
284 
    .await?;
285 
    for j in jobs {
286 
        sqlx::query(
287 
            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib, approval, environment, priority, queued_ms) \
288 
             VALUES ($1, $2, $3, $4, $5, $6, $7, 'queued', $8, $9, $10, $11, $12)",
289 
        )
290 
        .bind(id)
291 
        .bind(&j.name)
292 
        .bind(j.depth)
293 
        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))
294 
        .bind(&j.spec)
295 
        .bind(&j.sandbox)
296 
        .bind(&j.runtime)
297 
        .bind(j.need_disk_mib)
298 
        .bind(if needs_approval(&j.spec) { "waiting" } else { "" })
299 
        .bind(environment_of(&j.spec))
300 
        .bind(priority_of(&j.spec))
301 
        .bind(now)
302 
        .execute(&mut *tx)
303 
        .await?;
304 
    }
305 
    tx.commit().await?;
306 
    Ok(id)
307 
}
308
309 
/// The environment a compiled job deploys to, or empty (oxygenflow#59).
310 
fn environment_of(spec: &str) -> String {
311 
    serde_json::from_str::<serde_json::Value>(spec).ok().and_then(|v| v["environment"].as_str().map(String::from)).unwrap_or_default()
312 
}
313
314 
/// A compiled job's own priority (`low`, `normal`), or empty; anything
315 
/// else is not one a pipeline may give (oxygenhub#244).
316 
fn priority_of(spec: &str) -> String {
317 
    serde_json::from_str::<serde_json::Value>(spec)
318 
        .ok()
319 
        .and_then(|v| v["priority"].as_str().map(String::from))
320 
        .filter(|p| p == "low" || p == "normal")
321 
        .unwrap_or_default()
322 
}
323
324 
/// Whether a compiled job asks for approval (oxygenflow#57).
325 
fn needs_approval(spec: &str) -> bool {
326 
    serde_json::from_str::<serde_json::Value>(spec).is_ok_and(|v| !v["approval"].is_null())
327 
}
328
329 
/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).
330 
fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {
331 
    let mut out: Vec<String> = [spec, sandbox]
332 
        .iter()
333 
        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())
334 
        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())
335 
        .filter_map(|l| l.as_str().map(String::from))
336 
        .collect();
337 
    out.sort();
338 
    out.dedup();
339 
    out
340 
}
341
342 
impl Job {
343 
    /// The runner labels this job needs.
344 
    pub fn needs(&self) -> Vec<String> {
345 
        labels_in(&self.spec, &self.sandbox)
346 
    }
347
348 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
349 
    pub fn arch(&self) -> Option<String> {
350 
        arch_in(&self.sandbox)
351 
    }
352 
}
353
354 
/// The architecture a sandbox's JSON asks for, if any.
355 
fn arch_in(sandbox: &str) -> Option<String> {
356 
    serde_json::from_str::<serde_json::Value>(sandbox).ok()?["arch"].as_str().map(String::from)
357 
}
358
359 
/// A job waiting for or held by a runner, with where it comes from.
360 
#[derive(Clone, Debug, FromRow)]
361 
pub struct QueueEntry {
362 
    pub job_id: i64,
363 
    pub run_id: i64,
364 
    pub repo_id: i64,
365 
    pub owner: String,
366 
    pub name: String,
367 
    pub visibility: String,
368 
    pub run_number: i64,
369 
    pub job: String,
370 
    pub status: String,
371 
    /// The runner holding it, when running.
372 
    pub runner: String,
373 
    pub spec: String,
374 
    pub sandbox: String,
375 
    /// When it started running.
376 
    pub started_ms: Option<i64>,
377 
    /// Its level (0 urgent … 3 low, oxygenhub#244) and, when expedited, who
378 
    /// did it and why.
379 
    pub priority_rank: i64,
380 
    pub priority_note: String,
381 
}
382
383 
impl QueueEntry {
384 
    pub fn priority(&self) -> &'static str {
385 
        PRIORITIES.get(self.priority_rank as usize).copied().unwrap_or("normal")
386 
    }
387 
}
388
389 
impl QueueEntry {
390 
    /// The runner labels the job needs.
391 
    pub fn needs(&self) -> Vec<String> {
392 
        labels_in(&self.spec, &self.sandbox)
393 
    }
394
395 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
396 
    pub fn arch(&self) -> Option<String> {
397 
        arch_in(&self.sandbox)
398 
    }
399 
}
400
401 
/// Every job of a live run that is running or queued: running ones first,
402 
/// then queued ones in the order runners claim them (hub spec §13.2).
403 
pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
404 
    sqlx::query_as(&format!(
405 
        "SELECT j.id AS job_id, j.run_id, r.repo_id, rep.owner, rep.name, rep.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox, j.started_ms, \
406 
           CAST({PRIORITY_RANK} AS BIGINT) AS priority_rank, {PRIORITY_NOTE} AS priority_note \
407 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \
408 
         WHERE j.status IN ('queued', 'running') AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \
409 
         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, \
410 
           CASE j.status WHEN 'running' THEN 0 ELSE {PRIORITY_RANK} END, \
411 
           CASE j.status WHEN 'running' THEN 0 ELSE ROW_NUMBER() OVER (PARTITION BY r.repo_id, j.status ORDER BY j.run_id, j.depth, j.id) \
412 
           + (SELECT COUNT(*) FROM flow_jobs x JOIN flow_runs xr ON xr.id = x.run_id WHERE x.status = 'running' AND xr.repo_id = r.repo_id) END, \
413 
           j.run_id, j.depth, j.id"
414 
    ))
415 
    .fetch_all(pool)
416 
    .await
417 
}
418
419 
/// A repository's runs, newest first, numbered below `before`.
420 
pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
421 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND number < $2 ORDER BY number DESC LIMIT $3")
422 
        .bind(repo_id)
423 
        .bind(before.unwrap_or(i64::MAX))
424 
        .bind(limit)
425 
        .fetch_all(pool)
426 
        .await
427 
}
428
429 
/// The latest run on this bookmark of this repository, for what a push
430 
/// changed since (a job's scope).
431 
/// The run a push to `bookmark` is scoped against: the newest that was not
432 
/// cancelled, nor a push run still going (a new push supersedes it), so
433 
/// that what a cancelled run would have checked is checked by the next.
434 
pub async fn latest_on(pool: &Pool, repo_id: i64, bookmark: &str) -> sqlx::Result<Option<Run>> {
435 
    sqlx::query_as(
436 
        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND heads != '' AND status != 'cancelled'
437 
         AND NOT (trigger = 'push' AND status IN ('queued', 'running')) ORDER BY id DESC LIMIT 1",
438 
    )
439 
    .bind(repo_id)
440 
    .bind(bookmark)
441 
    .fetch_optional(pool)
442 
    .await
443 
}
444
445 
pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
446 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND id < $2 ORDER BY id DESC LIMIT $3")
447 
        .bind(repo_id)
448 
        .bind(before.unwrap_or(i64::MAX))
449 
        .bind(limit)
450 
        .fetch_all(pool)
451 
        .await
452 
}
453
454 
/// A repository's run by its number there.
455 
pub async fn run(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<Run>> {
456 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND number = $2").bind(repo_id).bind(number).fetch_optional(pool).await
457 
}
458
459 
pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {
460 
    sqlx::query_as("SELECT * FROM flow_runs WHERE id = $1").bind(id).fetch_optional(pool).await
461 
}
462
463 
pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {
464 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1 ORDER BY depth, name").bind(run_id).fetch_all(pool).await
465 
}
466
467 
pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {
468 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = $1").bind(id).fetch_optional(pool).await
469 
}
470
471 
/// The running job a job token belongs to, while it may still be used.
472 
pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {
473 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = $1 AND status = 'running' AND started_ms >= $2")
474 
        .bind(token_hash)
475 
        .bind(now_ms() - max_age_ms)
476 
        .fetch_optional(pool)
477 
        .await
478 
}
479
480 
/// Take the oldest queued job whose dependencies all passed and whose tier
481 
/// the runner offers, and mark it running for `runner` with a token.
482 
/// Let a job's token also read these repositories (its pinned subrepos).
483 
pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {
484 
    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();
485 
    sqlx::query("UPDATE flow_jobs SET also_read = $1 WHERE id = $2").bind(list.join(",")).bind(job_id).execute(pool).await?;
486 
    Ok(())
487 
}
488
489 
impl Job {
490 
    pub fn also_read_ids(&self) -> Vec<i64> {
491 
        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()
492 
    }
493 
}
494
495 
/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no
496 
/// limit, and the rootfs images it holds, `None` when it does not say.
497 
#[derive(Clone, Debug, Default)]
498 
pub struct Room {
499 
    pub memory_mib: Option<u32>,
500 
    pub cpus: Option<u32>,
501 
    pub images: Option<std::collections::HashSet<String>>,
502 
    /// The runner's architecture (`x86_64`, `riscv64`, …), empty when unknown.
503 
    pub arch: String,
504 
    /// Each image's architecture as runners have reported it (`flow_images`).
505 
    pub image_archs: std::collections::HashMap<String, String>,
506 
    /// Jobs this runner declined, not to be offered to it again (spec §11.5).
507 
    pub declined: std::collections::HashSet<i64>,
508 
    /// Images the hub holds that this runner would fetch when it lacks
509 
    /// them (oxygenflow#39); empty for a runner that does not fetch.
510 
    pub fetchable: std::collections::HashSet<String>,
511 
    /// Its free disk now, from its claim's facts (oxygenflow#40); `None`
512 
    /// when it does not say.
513 
    pub free_disk_mib: Option<u64>,
514 
}
515
516 
impl Room {
517 
    /// Whether a job needing `need` MiB of disk fits in its free disk.
518 
    pub fn disk_fits(&self, need: Option<i64>) -> bool {
519 
        match (need, self.free_disk_mib) {
520 
            (Some(n), Some(free)) => n.max(0) as u64 <= free,
521 
            _ => true,
522 
        }
523 
    }
524 
}
525
526 
impl Room {
527 
    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,
528 
    /// and its rootfs image among the runner's (oxygenhub#73).
529 
    pub fn fits(&self, sandbox_json: &str) -> bool {
530 
        let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();
531 
        let need = |k: &str| v[k].as_u64().unwrap_or(0);
532 
        let rootfs = v["rootfs"].as_str().unwrap_or("");
533 
        // The architecture: the sandbox's own ask, else the image's as read
534 
        // or learned from its passes (oxygenhub#173); one nobody has read
535 
        // or run yet matches (OxygenFlow spec §11.3).
536 
        let wanted = v["arch"].as_str().map(String::from).or_else(|| self.image_archs.get(rootfs).cloned());
537 
        let arch_ok = self.arch.is_empty() || wanted.as_deref().is_none_or(|w| w == self.arch);
538 
        arch_ok
539 
            && self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)
540 
            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)
541 
            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)) || self.fetchable.contains(rootfs))
542 
    }
543 
}
544
545 
/// Remember which architecture each image is for, as a runner reports it.
546 
pub async fn record_image_archs(pool: &Pool, archs: &std::collections::BTreeMap<String, String>) -> sqlx::Result<()> {
547 
    for (hash, arch) in archs.iter().take(500) {
548 
        if hash.len() != 64 || !flow_core::arch_ok(arch) {
549 
            continue;
550 
        }
551 
        sqlx::query("INSERT INTO flow_images (hash, arch, seen_ms) VALUES ($1, $2, $3) ON CONFLICT (hash) DO UPDATE SET arch = excluded.arch, seen_ms = excluded.seen_ms")
552 
        .bind(hash.to_ascii_lowercase())
553 
        .bind(arch)
554 
        .bind(now_ms())
555 
        .execute(pool)
556 
        .await?;
557 
    }
558 
    Ok(())
559 
}
560
561 
/// Give every image nobody has read the architecture of the runners its
562 
/// jobs passed on (oxygenhub#173), for the jobs before the hub learned at
563 
/// each pass. Never overrides a known architecture.
564 
pub async fn learn_archs_from_runs(pool: &Pool) -> sqlx::Result<u64> {
565 
    let rootfs = crate::db::json_text("j.sandbox", "rootfs");
566 
    let r = sqlx::query(&format!(
567 
        "INSERT INTO flow_images (hash, arch, seen_ms)
568 
         SELECT lower({rootfs}), MIN(r.arch), $1 FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id
569 
         WHERE j.status = 'passed' AND r.arch != '' AND length({rootfs}) = 64
570 
         GROUP BY lower({rootfs})
571 
         ON CONFLICT (hash) DO NOTHING"
572 
    ))
573 
    .bind(now_ms())
574 
    .execute(pool)
575 
    .await?;
576 
    Ok(r.rows_affected())
577 
}
578
579 
/// Every image's architecture the hub knows.
580 
pub async fn image_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
581 
    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, arch FROM flow_images").fetch_all(pool).await?;
582 
    Ok(rows.into_iter().collect())
583 
}
584
585 
pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {
586 
    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await
587 
}
588
589 
/// The first queued job a runner with these tiers, this room and this
590 
/// scope may take, now running under `runner`. Repositories take turns
591 
/// (fair share, oxygenhub#244): a job's turn is its place in its own
592 
/// repository's queue plus that repository's running jobs, lowest first,
593 
/// oldest first among equals, so one repository's burst can't hold up the
594 
/// rest. A scoped runner never sees
595 
/// another scope's jobs (oxygenhub#41).
596 
pub async fn claim_within(
597 
    pool: &Pool,
598 
    runtimes: &[String],
599 
    runner: &str,
600 
    token_hash: &str,
601 
    room: Room,
602 
    scope: &super::runners::Scope,
603 
    labels: &[String],
604 
) -> sqlx::Result<Option<Job>> {
605 
    claim_ranked(pool, runtimes, runner, token_hash, room, scope, labels, &|_| false).await
606 
}
607
608 
/// [`claim_within`], passing over a job `elsewhere` says a better runner
609 
/// should take (oxygenflow#42).
610 
#[allow(clippy::too_many_arguments)]
611 
pub async fn claim_ranked(
612 
    pool: &Pool,
613 
    runtimes: &[String],
614 
    runner: &str,
615 
    token_hash: &str,
616 
    room: Room,
617 
    scope: &super::runners::Scope,
618 
    labels: &[String],
619 
    elsewhere: &(dyn Fn(&Job) -> bool + Sync),
620 
) -> sqlx::Result<Option<Job>> {
621 
    let mut tx = super::begin_write(pool).await?;
622 
    let queued: Vec<Job> = sqlx::query_as(&format!(
623 
        "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 \
624 
         WHERE j.status = 'queued' AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \
625 
         AND ($1 = '' OR rep.owner = $2) AND ($3 IS NULL OR r.repo_id = $4) \
626 
         AND NOT ($5 = '' AND $6 IS NULL AND rep.flow = 'own') \
627 
         ORDER BY {PRIORITY_RANK}, ROW_NUMBER() OVER (PARTITION BY r.repo_id ORDER BY j.run_id, j.depth, j.id) \
628 
           + (SELECT COUNT(*) FROM flow_jobs x JOIN flow_runs xr ON xr.id = x.run_id WHERE x.status = 'running' AND xr.repo_id = r.repo_id), \
629 
           j.run_id, j.depth, j.id LIMIT 200"
630 
    ))
631 
    .bind(&scope.owner)
632 
    .bind(&scope.owner)
633 
    .bind(scope.repo_id)
634 
    .bind(scope.repo_id)
635 
    // A repository on its own runners only (#225): the site's runners skip it.
636 
    .bind(&scope.owner)
637 
    .bind(scope.repo_id)
638 
    .fetch_all(&mut *tx)
639 
    .await?;
640 
    for j in queued {
641 
        if !runtimes.iter().any(|r| r == &j.runtime)
642 
            || !room.fits(&j.sandbox)
643 
            || !room.disk_fits(j.need_disk_mib)
644 
            || room.declined.contains(&j.id)
645 
        {
646 
            continue;
647 
        }
648 
        // Every label the job and its sandbox ask for (`runs-on`).
649 
        if !j.needs().iter().all(|l| labels.contains(l)) {
650 
            continue;
651 
        }
652 
        let deps = j.depends_on();
653 
        // When it became ready: queued, its dependencies passed, approved.
654 
        let mut ready = j.queued_ms.unwrap_or(0).max(j.approved_ms.unwrap_or(0));
655 
        if !deps.is_empty() {
656 
            let mut passed = 0i64;
657 
            for d in &deps {
658 
                let st: Option<(String, Option<i64>)> =
659 
                    sqlx::query_as("SELECT status, finished_ms FROM flow_jobs WHERE run_id = $1 AND name = $2")
660 
                        .bind(j.run_id)
661 
                        .bind(d)
662 
                        .fetch_optional(&mut *tx)
663 
                        .await?;
664 
                if let Some((status, finished)) = st {
665 
                    if status == "passed" {
666 
                        passed += 1;
667 
                        ready = ready.max(finished.unwrap_or(0));
668 
                    }
669 
                }
670 
            }
671 
            if passed != deps.len() as i64 {
672 
                continue;
673 
            }
674 
        }
675 
        if elsewhere(&j) {
676 
            continue;
677 
        }
678 
        let now = now_ms();
679 
        let done = sqlx::query(
680 
            "UPDATE flow_jobs SET status = 'running', runner = $1, token_hash = $2, started_ms = $3, waiting = '', ready_ms = $5 \
681 
             WHERE id = $4 AND status = 'queued'",
682 
        )
683 
        .bind(runner)
684 
        .bind(token_hash)
685 
        .bind(now)
686 
        .bind(j.id)
687 
        .bind(if ready > 0 { ready.min(now) } else { now })
688 
        .execute(&mut *tx)
689 
        .await?;
690 
        if done.rows_affected() == 1 {
691 
            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = $1 AND status = 'queued'")
692 
                .bind(j.run_id)
693 
                .execute(&mut *tx)
694 
                .await?;
695 
            tx.commit().await?;
696 
            return job(pool, j.id).await;
697 
        }
698 
    }
699 
    tx.commit().await?;
700 
    Ok(None)
701 
}
702
703 
/// How long a decline keeps a job from its runner when the runner's
704 
/// profile stays the same: a reason such as memory in use passes.
705 
pub const DECLINE_HOLDS_MS: i64 = 10 * 60_000;
706
707 
/// The jobs runner `runner_id` declined with this profile, recently.
708 
pub async fn declined_by(pool: &Pool, runner_id: i64, profile: &str) -> sqlx::Result<std::collections::HashSet<i64>> {
709 
    let rows: Vec<(i64,)> = sqlx::query_as("SELECT job_id FROM flow_declines WHERE runner_id = $1 AND profile = $2 AND ms > $3")
710 
        .bind(runner_id)
711 
        .bind(profile)
712 
        .bind(now_ms() - DECLINE_HOLDS_MS)
713 
        .fetch_all(pool)
714 
        .await?;
715 
    Ok(rows.into_iter().map(|(j,)| j).collect())
716 
}
717
718 
/// A decline: one runner's reason for handing a job back.
719 
#[derive(Clone, Debug, sqlx::FromRow)]
720 
pub struct DeclineRow {
721 
    pub job_id: i64,
722 
    pub runner_id: i64,
723 
    pub runner: String,
724 
    pub reason: String,
725 
    pub ms: i64,
726 
}
727
728 
/// Runner `runner_id` hands back job `job_id`, running on it: the decline
729 
/// is kept (one per job and runner, the latest), and the job is queued
730 
/// again. Returns every decline the job has had, or `None` when the job
731 
/// was not running on that runner.
732 
pub async fn decline(
733 
    pool: &Pool,
734 
    job_id: i64,
735 
    runner_id: i64,
736 
    runner: &str,
737 
    profile: &str,
738 
    reason: &str,
739 
) -> sqlx::Result<Option<Vec<DeclineRow>>> {
740 
    let mut tx = super::begin_write(pool).await?;
741 
    let back = sqlx::query(
742 
        "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, runner_id = NULL \
743 
         WHERE id = $1 AND status = 'running' AND runner_id = $2",
744 
    )
745 
    .bind(job_id)
746 
    .bind(runner_id)
747 
    .execute(&mut *tx)
748 
    .await?;
749 
    if back.rows_affected() != 1 {
750 
        tx.rollback().await?;
751 
        return Ok(None);
752 
    }
753 
    sqlx::query(
754 
        "INSERT INTO flow_declines (job_id, runner_id, runner, profile, reason, ms) VALUES ($1, $2, $3, $4, $5, $6) \
755 
         ON CONFLICT (job_id, runner_id) DO UPDATE SET runner = excluded.runner, profile = excluded.profile, reason = excluded.reason, ms = excluded.ms",
756 
    )
757 
    .bind(job_id)
758 
    .bind(runner_id)
759 
    .bind(runner)
760 
    .bind(profile)
761 
    .bind(reason)
762 
    .bind(now_ms())
763 
    .execute(&mut *tx)
764 
    .await?;
765 
    let all: Vec<DeclineRow> =
766 
        sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = $1 ORDER BY ms")
767 
            .bind(job_id)
768 
            .fetch_all(&mut *tx)
769 
            .await?;
770 
    tx.commit().await?;
771 
    Ok(Some(all))
772 
}
773
774 
/// The declines of job `job_id`, oldest first.
775 
pub async fn declines_of_job(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<DeclineRow>> {
776 
    sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = $1 ORDER BY ms")
777 
        .bind(job_id)
778 
        .fetch_all(pool)
779 
        .await
780 
}
781
782 
/// A runner's latest declines, newest first, with where the job is.
783 
#[derive(Clone, Debug, sqlx::FromRow)]
784 
pub struct RunnerDecline {
785 
    pub job_id: i64,
786 
    pub job: String,
787 
    pub run_number: i64,
788 
    pub owner: String,
789 
    pub name: String,
790 
    pub reason: String,
791 
    pub ms: i64,
792 
}
793
794 
pub async fn declines_of_runner(pool: &Pool, runner_id: i64, limit: i64) -> sqlx::Result<Vec<RunnerDecline>> {
795 
    sqlx::query_as(
796 
        "SELECT d.job_id, j.name AS job, r.number AS run_number, rep.owner, rep.name, d.reason, d.ms \
797 
         FROM flow_declines d JOIN flow_jobs j ON j.id = d.job_id JOIN flow_runs r ON r.id = j.run_id \
798 
         JOIN repositories rep ON rep.id = r.repo_id WHERE d.runner_id = $1 ORDER BY d.ms DESC LIMIT $2",
799 
    )
800 
    .bind(runner_id)
801 
    .bind(limit)
802 
    .fetch_all(pool)
803 
    .await
804 
}
805
806 
/// Queued jobs whose dependencies have passed, in their live runs, with
807 
/// their repositories (owner, id): the ones a runner could take now.
808 
pub async fn ready_queued(pool: &Pool) -> sqlx::Result<Vec<(Job, String, i64)>> {
809 
    let queued: Vec<Job> = sqlx::query_as(
810 
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
811 
         WHERE j.status = 'queued' AND j.approval != 'waiting' AND r.status IN ('queued', 'running') ORDER BY j.run_id, j.depth, j.id LIMIT 500",
812 
    )
813 
    .fetch_all(pool)
814 
    .await?;
815 
    let mut out = Vec::new();
816 
    for j in queued {
817 
        let mut ready = true;
818 
        for d in j.depends_on() {
819 
            let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = $1 AND name = $2")
820 
                .bind(j.run_id)
821 
                .bind(&d)
822 
                .fetch_optional(pool)
823 
                .await?;
824 
            if st.as_deref() != Some("passed") {
825 
                ready = false;
826 
                break;
827 
            }
828 
        }
829 
        if !ready {
830 
            continue;
831 
        }
832 
        let repo: Option<(String, i64)> =
833 
            sqlx::query_as("SELECT rep.owner, rep.id FROM flow_runs r JOIN repositories rep ON rep.id = r.repo_id WHERE r.id = $1")
834 
                .bind(j.run_id)
835 
                .fetch_optional(pool)
836 
                .await?;
837 
        if let Some((owner, id)) = repo {
838 
            out.push((j, owner, id));
839 
        }
840 
    }
841 
    Ok(out)
842 
}
843
844 
pub async fn set_waiting(pool: &Pool, job_id: i64, waiting: &str) -> sqlx::Result<()> {
845 
    sqlx::query("UPDATE flow_jobs SET waiting = $1 WHERE id = $2 AND waiting <> $3")
846 
        .bind(waiting)
847 
        .bind(job_id)
848 
        .bind(waiting)
849 
        .execute(pool)
850 
        .await?;
851 
    Ok(())
852 
}
853
854 
/// Append records; returns how many were kept (the rest exceed the cap).
855 
pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {
856 
    let mut tx = super::begin_write(pool).await?;
857 
    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = $1").bind(job_id).fetch_one(&mut *tx).await?;
858 
    let room = (cap - count).max(0) as usize;
859 
    let keep = &records[..records.len().min(room)];
860 
    for (i, r) in keep.iter().enumerate() {
861 
        sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES ($1, $2, $3, $4, $5)")
862 
            .bind(job_id)
863 
            .bind(count + i as i64 + 1)
864 
            .bind(r.t as i64)
865 
            .bind(&r.stream)
866 
            .bind(&r.text)
867 
            .execute(&mut *tx)
868 
            .await?;
869 
    }
870 
    sqlx::query("UPDATE flow_jobs SET log_count = log_count + $1 WHERE id = $2")
871 
        .bind(keep.len() as i64)
872 
        .bind(job_id)
873 
        .execute(&mut *tx)
874 
        .await?;
875 
    tx.commit().await?;
876 
    Ok(keep.len())
877 
}
878
879 
pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {
880 
    sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = $1 AND seq > $2 ORDER BY seq LIMIT $3")
881 
        .bind(job_id)
882 
        .bind(after)
883 
        .bind(limit)
884 
        .fetch_all(pool)
885 
        .await
886 
}
887
888 
/// Jobs running since before `ms` (the oldest first).
889 
/// Jobs still running at `now` past their own time limit (`default_s` for
890 
/// one claimed without) plus `grace_ms`.
891 
pub async fn running_past_limit(pool: &Pool, now: i64, default_s: u64, grace_ms: i64) -> sqlx::Result<Vec<Job>> {
892 
    sqlx::query_as(
893 
        "SELECT * FROM flow_jobs WHERE status = 'running' \
894 
         AND COALESCE(started_ms, 0) + (CASE WHEN timeout_s > 0 THEN timeout_s ELSE $1 END) * 1000 + $2 < $3 ORDER BY started_ms",
895 
    )
896 
    .bind(default_s as i64)
897 
    .bind(grace_ms)
898 
    .bind(now)
899 
    .fetch_all(pool)
900 
    .await
901 
}
902
903 
/// Record the limit a claimed job runs under.
904 
pub async fn set_job_timeout(pool: &Pool, job_id: i64, timeout_s: u64) -> sqlx::Result<()> {
905 
    sqlx::query("UPDATE flow_jobs SET timeout_s = $1 WHERE id = $2").bind(timeout_s as i64).bind(job_id).execute(pool).await?;
906 
    Ok(())
907 
}
908
909 
pub async fn running_since_before(pool: &Pool, ms: i64) -> sqlx::Result<Vec<Job>> {
910 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < $1 ORDER BY started_ms")
911 
        .bind(ms)
912 
        .fetch_all(pool)
913 
        .await
914 
}
915
916 
/// The jobs running on the runner with this id.
917 
pub async fn running_on(pool: &Pool, runner_id: i64) -> sqlx::Result<Vec<Job>> {
918 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = $1 ORDER BY id").bind(runner_id).fetch_all(pool).await
919 
}
920
921 
pub struct Outcome<'a> {
922 
    pub status: &'a str,
923 
    pub exit_code: Option<i64>,
924 
    pub error: &'a str,
925 
    pub receipt: &'a str,
926 
    pub receipt_hash: &'a str,
927 
}
928
929 
/// Record a job's end, skip the jobs that can no longer run, and settle the
930 
/// run. Returns the jobs whose status changed (for their checks).
931 
pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {
932 
    let mut tx = super::begin_write(pool).await?;
933 
    let now = now_ms();
934 
    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = $1").bind(job_id).fetch_one(&mut *tx).await?;
935 
    sqlx::query("UPDATE flow_jobs SET status = $1, exit_code = $2, error = $3, receipt = $4, receipt_hash = $5, finished_ms = $6, token_hash = '' WHERE id = $7")
936 
        .bind(o.status)
937 
        .bind(o.exit_code)
938 
        .bind(o.error)
939 
        .bind(o.receipt)
940 
        .bind(o.receipt_hash)
941 
        .bind(now)
942 
        .bind(job_id)
943 
        .execute(&mut *tx)
944 
        .await?;
945 
    // A pass teaches what its image is for, when nobody has read it
946 
    // (oxygenhub#173): the architecture of the runner it passed on.
947 
    if o.status == "passed" {
948 
        let rootfs = serde_json::from_str::<serde_json::Value>(&j.sandbox)
949 
            .ok()
950 
            .and_then(|v| v["rootfs"].as_str().map(str::to_ascii_lowercase))
951 
            .unwrap_or_default();
952 
        if rootfs.len() == 64 {
953 
            sqlx::query(
954 
                "INSERT INTO flow_images (hash, arch, seen_ms) SELECT $1, r.arch, $2 FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id \
955 
                 WHERE j.id = $3 AND r.arch != '' ON CONFLICT (hash) DO NOTHING",
956 
            )
957 
            .bind(&rootfs)
958 
            .bind(now)
959 
            .bind(job_id)
960 
            .execute(&mut *tx)
961 
            .await?;
962 
        }
963 
    }
964 
    let mut changed = vec![job_id];
965 
    // Dependents of anything that did not pass are skipped, transitively.
966 
    loop {
967 
        let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1").bind(j.run_id).fetch_all(&mut *tx).await?;
968 
        let blocked: Vec<i64> = all
969 
            .iter()
970 
            .filter(|x| x.status == "queued")
971 
            .filter(|x| {
972 
                x.depends_on()
973 
                    .iter()
974 
                    .any(|d| all.iter().any(|y| &y.name == d && matches!(y.status.as_str(), "failed" | "skipped" | "error" | "cancelled")))
975 
            })
976 
            .map(|x| x.id)
977 
            .collect();
978 
        if blocked.is_empty() {
979 
            break;
980 
        }
981 
        for id in blocked {
982 
            sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = $1 WHERE id = $2")
983 
                .bind(now)
984 
                .bind(id)
985 
                .execute(&mut *tx)
986 
                .await?;
987 
            changed.push(id);
988 
        }
989 
    }
990 
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1").bind(j.run_id).fetch_all(&mut *tx).await?;
991 
    if all.iter().all(Job::done) {
992 
        let status = if all.iter().all(|x| x.status == "passed") { "passed" } else { "failed" };
993 
        sqlx::query("UPDATE flow_runs SET status = $1, finished_ms = $2 WHERE id = $3")
994 
            .bind(status)
995 
            .bind(now)
996 
            .bind(j.run_id)
997 
            .execute(&mut *tx)
998 
            .await?;
999 
    }
1000 
    tx.commit().await?;
1001 
    let mut out = Vec::new();
1002 
    for id in changed {
1003 
        if let Some(x) = job(pool, id).await? {
1004 
            out.push(x);
1005 
        }
1006 
    }
1007 
    Ok(out)
1008 
}
1009
1010 
/// Cancel a run (oxygenflow#32): its jobs not yet done end as cancelled,
1011 
/// and so does the run. A runner learns of it when the hub next refuses
1012 
/// its log lines, and stops the job. Returns the jobs whose status changed.
1013 
pub async fn cancel_run(pool: &Pool, run_id: i64, why: &str) -> sqlx::Result<Vec<Job>> {
1014 
    let now = now_ms();
1015 
    let mut tx = pool.begin().await?;
1016 
    let ids: Vec<(i64,)> = sqlx::query_as("SELECT id FROM flow_jobs WHERE run_id = $1 AND status IN ('queued', 'running')")
1017 
        .bind(run_id)
1018 
        .fetch_all(&mut *tx)
1019 
        .await?;
1020 
    sqlx::query(
1021 
        "UPDATE flow_jobs SET status = 'cancelled', error = $1, finished_ms = $2, token_hash = ''
1022 
         WHERE run_id = $3 AND status IN ('queued', 'running')",
1023 
    )
1024 
    .bind(why)
1025 
    .bind(now)
1026 
    .bind(run_id)
1027 
    .execute(&mut *tx)
1028 
    .await?;
1029 
    sqlx::query("UPDATE flow_runs SET status = 'cancelled', finished_ms = $1 WHERE id = $2 AND status IN ('queued', 'running')")
1030 
        .bind(now)
1031 
        .bind(run_id)
1032 
        .execute(&mut *tx)
1033 
        .await?;
1034 
    tx.commit().await?;
1035 
    let mut out = Vec::new();
1036 
    for (id,) in ids {
1037 
        if let Some(x) = job(pool, id).await? {
1038 
            out.push(x);
1039 
        }
1040 
    }
1041 
    Ok(out)
1042 
}
1043
1044 
/// Run job `name` of finished run `run_id` again, with every job that
1045 
/// depends on it (oxygenflow#31): back to the queue as if new, their logs,
1046 
/// artifacts and reports gone, and the run running again. The jobs they
1047 
/// depend on keep their results. The jobs reset; none when there is no
1048 
/// such job.
1049 
pub async fn rerun_job(pool: &Pool, run_id: i64, name: &str) -> sqlx::Result<Vec<Job>> {
1050 
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1").bind(run_id).fetch_all(pool).await?;
1051 
    if !all.iter().any(|j| j.name == name) {
1052 
        return Ok(Vec::new());
1053 
    }
1054 
    let mut again = vec![name.to_string()];
1055 
    loop {
1056 
        let more: Vec<String> = all
1057 
            .iter()
1058 
            .filter(|j| !again.contains(&j.name) && j.depends_on().iter().any(|d| again.contains(d)))
1059 
            .map(|j| j.name.clone())
1060 
            .collect();
1061 
        if more.is_empty() {
1062 
            break;
1063 
        }
1064 
        again.extend(more);
1065 
    }
1066 
    let ids: Vec<i64> = all.iter().filter(|j| again.contains(&j.name)).map(|j| j.id).collect();
1067 
    let mut tx = pool.begin().await?;
1068 
    for id in &ids {
1069 
        for table in ["flow_logs", "flow_artifacts", "flow_declines", "test_reports", "coverage_reports", "reports"] {
1070 
            sqlx::query(&format!("DELETE FROM {table} WHERE job_id = $1")).bind(id).execute(&mut *tx).await?;
1071 
        }
1072 
        sqlx::query(
1073 
            "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, finished_ms = NULL,
1074 
             exit_code = NULL, error = '', receipt = '', receipt_hash = '', log_count = 0, also_read = '', receipt_attest = '',
1075 
             receipt_signer = '', receipt_note = '', runner_id = NULL, waiting = '', placed = '', peak_disk_mib = NULL,
1076 
             peak_memory_mib = NULL, approval = CASE WHEN approval = '' THEN '' ELSE 'waiting' END, approved_by = '',
1077 
             approved_ms = NULL, approval_seal = '', approved_via = '', queued_ms = $2, ready_ms = NULL WHERE id = $1",
1078 
        )
1079 
        .bind(id)
1080 
        .bind(now_ms())
1081 
        .execute(&mut *tx)
1082 
        .await?;
1083 
    }
1084 
    sqlx::query("UPDATE flow_runs SET status = 'running', finished_ms = NULL WHERE id = $1").bind(run_id).execute(&mut *tx).await?;
1085 
    tx.commit().await?;
1086 
    let mut out = Vec::new();
1087 
    for id in ids {
1088 
        if let Some(x) = job(pool, id).await? {
1089 
            out.push(x);
1090 
        }
1091 
    }
1092 
    Ok(out)
1093 
}
1094
1095 
/// The push runs of `bookmark` still going, older than run `number`.
1096 
pub async fn live_push_runs_before(pool: &Pool, repo_id: i64, bookmark: &str, number: i64) -> sqlx::Result<Vec<Run>> {
1097 
    sqlx::query_as(
1098 
        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND number < $3 AND trigger = 'push'
1099 
         AND status IN ('queued', 'running') ORDER BY id",
1100 
    )
1101 
    .bind(repo_id)
1102 
    .bind(bookmark)
1103 
    .bind(number)
1104 
    .fetch_all(pool)
1105 
    .await
1106 
}
1107
1108 
/// The newest run of exactly this state (heads joined with `+`, sorted).
1109 
pub async fn latest_for_heads(pool: &Pool, repo_id: i64, heads: &str) -> sqlx::Result<Option<Run>> {
1110 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND heads = $2 ORDER BY id DESC LIMIT 1")
1111 
        .bind(repo_id)
1112 
        .bind(heads)
1113 
        .fetch_optional(pool)
1114 
        .await
1115 
}
1116
1117 
/// Record what became of job `job_id`'s signed receipt.
1118 
pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {
1119 
    sqlx::query("UPDATE flow_jobs SET receipt_attest = $1, receipt_signer = $2, receipt_note = $3 WHERE id = $4")
1120 
        .bind(attest)
1121 
        .bind(signer)
1122 
        .bind(note)
1123 
        .bind(job_id)
1124 
        .execute(pool)
1125 
        .await?;
1126 
    Ok(())
1127 
}
1128
1129 
/// A job's artifact, kept as a blob of the run's repository.
1130 
#[derive(Clone, Debug, FromRow)]
1131 
pub struct Artifact {
1132 
    pub job_id: i64,
1133 
    pub name: String,
1134 
    pub path: String,
1135 
    pub blob: String,
1136 
    pub size: i64,
1137 
    /// BLAKE3 of the bytes.
1138 
    pub content: String,
1139 
    pub created_ms: i64,
1140 
}
1141
1142 
pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {
1143 
    sqlx::query(
1144 
        "INSERT INTO flow_artifacts (job_id, name, path, blob, size, content, created_ms) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (job_id, name) DO UPDATE SET path = excluded.path, blob = excluded.blob, size = excluded.size, content = excluded.content, created_ms = excluded.created_ms",
1145 
    )
1146 
    .bind(a.job_id)
1147 
    .bind(&a.name)
1148 
    .bind(&a.path)
1149 
    .bind(&a.blob)
1150 
    .bind(a.size)
1151 
    .bind(&a.content)
1152 
    .bind(a.created_ms)
1153 
    .execute(pool)
1154 
    .await?;
1155 
    Ok(())
1156 
}
1157
1158 
/// Every artifact blob a repository's runs keep, for garbage collection to
1159 
/// keep too (#126): the database is the only thing that names them.
1160 
pub async fn artifact_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
1161 
    sqlx::query_scalar(
1162 
        "SELECT DISTINCT a.blob 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 = $1",
1163 
    )
1164 
    .bind(repo_id)
1165 
    .fetch_all(pool)
1166 
    .await
1167 
}
1168
1169 
/// Job `job_id`'s artifacts, by name.
1170 
pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
1171 
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = $1 ORDER BY name").bind(job_id).fetch_all(pool).await
1172 
}
1173
1174 
/// An artifact with the run and job that made it, for a repository's list
1175 
/// of every run's outputs (oxygenhub#254).
1176 
#[derive(Clone, Debug, FromRow)]
1177 
pub struct RepoArtifact {
1178 
    pub run_number: i64,
1179 
    pub bookmark: String,
1180 
    pub job: String,
1181 
    pub name: String,
1182 
    pub path: String,
1183 
    pub size: i64,
1184 
    pub content: String,
1185 
    pub created_ms: i64,
1186 
}
1187
1188 
/// A repository's newest artifacts, at most `limit`.
1189 
pub async fn repo_artifacts(pool: &Pool, repo_id: i64, limit: i64) -> sqlx::Result<Vec<RepoArtifact>> {
1190 
    sqlx::query_as(
1191 
        "SELECT r.number AS run_number, r.bookmark, j.name AS job, a.name, a.path, a.size, a.content, a.created_ms
1192 
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id
1193 
         WHERE r.repo_id = $1 ORDER BY a.created_ms DESC, r.number DESC, j.name, a.name LIMIT $2",
1194 
    )
1195 
    .bind(repo_id)
1196 
    .bind(limit)
1197 
    .fetch_all(pool)
1198 
    .await
1199 
}
1200
1201 
/// Where an artifact with given content came from: the run and job that
1202 
/// built it, and the job's signed receipt when one was kept.
1203 
#[derive(Clone, Debug, FromRow)]
1204 
pub struct Provenance {
1205 
    pub run_number: i64,
1206 
    pub job: String,
1207 
    pub name: String,
1208 
    pub receipt_attest: String,
1209 
    pub receipt_signer: String,
1210 
}
1211
1212 
/// The build of the bytes whose BLAKE3 is `content` in repository
1213 
/// `repo_id`: a passed job, one with a kept signed receipt first, newest
1214 
/// run first.
1215 
pub async fn provenance(pool: &Pool, repo_id: i64, content: &str) -> sqlx::Result<Option<Provenance>> {
1216 
    sqlx::query_as(
1217 
        "SELECT r.number AS run_number, j.name AS job, a.name AS name, j.receipt_attest, j.receipt_signer \
1218 
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id \
1219 
         WHERE r.repo_id = $1 AND a.content = $2 AND j.status = 'passed' \
1220 
         ORDER BY (j.receipt_attest != '') DESC, r.number DESC LIMIT 1",
1221 
    )
1222 
    .bind(repo_id)
1223 
    .bind(content)
1224 
    .fetch_optional(pool)
1225 
    .await
1226 
}
1227
1228 
/// Approve job `id` if it is still waiting for a person (oxygenflow#57);
1229 
/// whether it was.
1230 
pub async fn approve_job(pool: &Pool, id: i64, who: &str, seal: &str, via: &str) -> sqlx::Result<bool> {
1231 
    let r = sqlx::query(
1232 
        "UPDATE flow_jobs SET approval = 'approved', approved_by = $1, approved_ms = $2, approval_seal = $3, approved_via = $5 \
1233 
         WHERE id = $4 AND status = 'queued' AND approval = 'waiting'",
1234 
    )
1235 
    .bind(who)
1236 
    .bind(now_ms())
1237 
    .bind(seal)
1238 
    .bind(id)
1239 
    .bind(via)
1240 
    .execute(pool)
1241 
    .await?;
1242 
    Ok(r.rows_affected() > 0)
1243 
}
1244
1245 
/// Note who rejected job `id` (it is then finished as cancelled).
1246 
pub async fn note_rejection(pool: &Pool, id: i64, who: &str, via: &str) -> sqlx::Result<()> {
1247 
    sqlx::query("UPDATE flow_jobs SET approved_by = $1, approved_ms = $2, approved_via = $4 WHERE id = $3")
1248 
        .bind(who)
1249 
        .bind(now_ms())
1250 
        .bind(id)
1251 
        .bind(via)
1252 
        .execute(pool)
1253 
        .await?;
1254 
    Ok(())
1255 
}
1256
1257 
/// Jobs waiting for a person, in runs still going.
1258 
pub async fn waiting_for_approval(pool: &Pool) -> sqlx::Result<Vec<Job>> {
1259 
    sqlx::query_as(
1260 
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
1261 
         WHERE j.status = 'queued' AND j.approval = 'waiting' AND r.status IN ('queued', 'running')",
1262 
    )
1263 
    .fetch_all(pool)
1264 
    .await
1265 
}
1266
1267 
/// One deployment: a job for an environment that passed.
1268 
#[derive(Clone, Debug, FromRow)]
1269 
pub struct Deployment {
1270 
    pub environment: String,
1271 
    pub job: String,
1272 
    pub run_number: i64,
1273 
    pub state: String,
1274 
    pub bookmark: String,
1275 
    pub actor: String,
1276 
    pub approved_by: String,
1277 
    pub finished_ms: i64,
1278 
}
1279
1280 
/// A repository's deployments, newest first (oxygenflow#59).
1281 
pub async fn deployments(pool: &Pool, repo_id: i64, limit: i64) -> sqlx::Result<Vec<Deployment>> {
1282 
    sqlx::query_as(
1283 
        "SELECT j.environment, j.name AS job, r.number AS run_number, r.state, r.bookmark, r.actor, j.approved_by, \
1284 
         COALESCE(j.finished_ms, 0) AS finished_ms FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
1285 
         WHERE r.repo_id = $1 AND j.environment != '' AND j.status = 'passed' ORDER BY j.finished_ms DESC LIMIT $2",
1286 
    )
1287 
    .bind(repo_id)
1288 
    .bind(limit)
1289 
    .fetch_all(pool)
1290 
    .await
1291 
}