OXYGENHUB
◆ v0.9.20 oxygenhub / crates / hub / src / db / flow.rs
63.5 KiB · Rust · last changed Runner updates in waves: a later wave takes a version once the waves before it proved it; an error stops it (refs andymccall/oxygenflow#84) 8 days ago source blame history raw
42fa9554 Andy McCall ↶ 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,
a0d4bbfb Andy McCall ↶ 10 
    /// The run's number within its repository, from 1 (migration 0019).
11 
    #[sqlx(default)]
12 
    pub number: i64,
a9f1c0bf Andy McCall ↶ 13 
    /// `push`, `proposal`, `tag:NAME`, `manual`, `schedule` or
14 
    /// `rerun:N:EVENT` (migration 0020).
1edac4b6 Andy McCall ↶ 15 
    #[sqlx(default)]
16 
    pub trigger: String,
42fa9554 Andy McCall ↶ 17 
    pub bookmark: String,
18 
    pub state: String,
19 
    pub heads: String,
20 
    pub pipeline_hash: String,
21 
    pub status: String,
22 
    pub error: String,
23 
    pub actor: String,
24 
    pub created_ms: i64,
25 
    pub finished_ms: Option<i64>,
107eaba5 Andy McCall ↶ 26 
    /// `urgent` when expedited (oxygenhub#244), else empty: the run follows
27 
    /// its repository's, organisation's or the site's level.
28 
    #[sqlx(default)]
29 
    pub priority: String,
30 
    /// Who expedited it and why.
31 
    #[sqlx(default)]
32 
    pub priority_note: String,
33 
}
34
35 
/// The priority levels, most urgent first (oxygenhub#244).
36 
pub const PRIORITIES: [&str; 4] = ["urgent", "high", "normal", "low"];
37
53a0fb2e Andy McCall ↶ 38 
/// SQL for the level a run's jobs get from where they are, as a rank
39 
/// (0 urgent … 3 low): the run's own (expedited), else its repository's,
40 
/// its organisation's, the site's (`settings.flow_priority`), else normal.
41 
macro_rules! level_rank {
42 
    () => {
43 
        "CASE COALESCE(NULLIF(r.priority, ''), NULLIF(rep.flow_priority, ''), \
44 
           NULLIF((SELECT o.flow_priority FROM organisations o WHERE o.name = rep.owner), ''), \
45 
           NULLIF((SELECT s.value FROM settings s WHERE s.key = 'flow_priority'), ''), 'normal') \
46 
         WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'low' THEN 3 ELSE 2 END"
47 
    };
48 
}
49
50 
/// SQL for a queued job's level as a rank (0 urgent … 3 low): the level
51 
/// its run gets from where it is, lowered by the job's own priority from
52 
/// its pipeline (`low`, `normal`), never raised by it; an expedited run is
53 
/// urgent whatever its jobs say (oxygenhub#244). Needs `j` (the job), `r`
54 
/// (its run) and `rep` (its repository) in the query.
55 
pub const PRIORITY_RANK: &str = concat!(
56 
    "CASE WHEN r.priority <> '' THEN ",
57 
    level_rank!(),
58 
    " WHEN j.priority = 'low' THEN 3 WHEN j.priority = 'normal' AND (",
59 
    level_rank!(),
60 
    ") < 2 THEN 2 ELSE ",
61 
    level_rank!(),
62 
    " END"
63 
);
64
65 
/// SQL for why a job stands where it does, beside [`PRIORITY_RANK`]: an
66 
/// expedited run's note, else the pipeline's word for a job it lowered.
67 
pub const PRIORITY_NOTE: &str =
68 
    "CASE WHEN r.priority <> '' THEN r.priority_note WHEN j.priority <> '' THEN 'pipeline: ' || j.priority ELSE r.priority_note END";
107eaba5 Andy McCall ↶ 69
76f93eae Andy McCall ↶ 70 
/// How many runs of `owner`'s repositories are expedited and still live.
71 
pub async fn expedited_live(pool: &Pool, owner: &str) -> sqlx::Result<i64> {
72 
    sqlx::query_scalar(
73 
        "SELECT COUNT(*) FROM flow_runs r JOIN repositories rep ON rep.id = r.repo_id \
74 
         WHERE rep.owner = $1 AND r.priority = 'urgent' AND r.status IN ('queued', 'running')",
75 
    )
76 
    .bind(owner)
77 
    .fetch_one(pool)
78 
    .await
79 
}
80
107eaba5 Andy McCall ↶ 81 
/// Raise run `run_id` to urgent, saying who and why; a run already ended
82 
/// is left alone. Whether it changed.
83 
pub async fn expedite(pool: &Pool, run_id: i64, note: &str) -> sqlx::Result<bool> {
84 
    let r = sqlx::query("UPDATE flow_runs SET priority = 'urgent', priority_note = $1 WHERE id = $2 AND status IN ('queued', 'running')")
85 
        .bind(note)
86 
        .bind(run_id)
87 
        .execute(pool)
88 
        .await?;
89 
    Ok(r.rows_affected() > 0)
42fa9554 Andy McCall ↶ 90 
}
91
92 
impl Run {
1edac4b6 Andy McCall ↶ 93 
    /// How the run page says what started it.
94 
    pub fn started_by(&self) -> String {
95 
        match self.trigger.as_str() {
96 
            "manual" => format!("started by {}", self.actor),
c889911b Andy McCall ↶ 97 
            "schedule" => "started on schedule".to_string(),
98 
            t if t.starts_with("tag:") => format!("tag {} pushed by {}", &t[4..], self.actor),
99 
            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", self.rerun_of().unwrap_or(""), self.actor),
1edac4b6 Andy McCall ↶ 100 
            _ => format!("pushed by {}", self.actor),
101 
        }
102 
    }
103
c889911b Andy McCall ↶ 104 
    /// The run a rerun repeats: `N` of `rerun:N` or `rerun:N:EVENT`.
105 
    pub fn rerun_of(&self) -> Option<&str> {
106 
        let rest = self.trigger.strip_prefix("rerun:")?;
107 
        Some(rest.split(':').next().unwrap_or(rest))
108 
    }
109
110 
    /// The trigger as people read it: `rerun:N`, without the event it repeats.
111 
    pub fn trigger_label(&self) -> String {
112 
        match self.rerun_of() {
113 
            Some(n) => format!("rerun:{n}"),
114 
            None => self.trigger.clone(),
115 
        }
116 
    }
117
42fa9554 Andy McCall ↶ 118 
    pub fn head_list(&self) -> Vec<&str> {
119 
        self.heads.split('+').filter(|h| !h.is_empty()).collect()
120 
    }
121
122 
    pub fn live(&self) -> bool {
123 
        matches!(self.status.as_str(), "queued" | "running")
124 
    }
125 
}
126
4d2e44d7 Andy McCall ↶ 127 
#[derive(Clone, Debug, Default, FromRow)]
42fa9554 Andy McCall ↶ 128 
pub struct Job {
129 
    pub id: i64,
130 
    pub run_id: i64,
131 
    pub name: String,
132 
    pub depth: i64,
133 
    pub depends: String,
134 
    pub spec: String,
135 
    pub sandbox: String,
136 
    pub runtime: String,
137 
    pub status: String,
138 
    pub runner: String,
139 
    pub token_hash: String,
140 
    pub started_ms: Option<i64>,
141 
    pub finished_ms: Option<i64>,
142 
    pub exit_code: Option<i64>,
143 
    pub error: String,
144 
    pub receipt: String,
145 
    pub receipt_hash: String,
146 
    pub log_count: i64,
fd3db644 Andy McCall ↶ 147 
    /// Comma-separated ids of other repositories the job token may read.
148 
    pub also_read: String,
cc19697f Andy McCall ↶ 149 
    /// The receipt attestation kept in the repository (Oxygen spec §14.6).
150 
    pub receipt_attest: String,
151 
    /// Who signed it: the runner's identity name.
152 
    pub receipt_signer: String,
153 
    /// Why a signed receipt was not kept.
154 
    pub receipt_note: String,
4d2e44d7 Andy McCall ↶ 155 
    /// Why a queued job no runner can take waits (spec §11.4), from the
156 
    /// last placement pass; empty when it is only waiting its turn.
157 
    pub waiting: String,
44a3d2eb Andy McCall ↶ 158 
    /// What it took, as its runner reported (oxygenflow#41), in MiB.
159 
    pub peak_disk_mib: Option<i64>,
160 
    pub peak_memory_mib: Option<i64>,
04683947 Andy McCall ↶ 161 
    /// Where its time went, JSON of phase to milliseconds (oxygenflow#35).
162 
    #[sqlx(default)]
163 
    pub phases: String,
44a3d2eb Andy McCall ↶ 164 
    /// The free disk it needs, learned from past runs of the same job.
165 
    pub need_disk_mib: Option<i64>,
501d8b26 Andy McCall ↶ 166 
    /// Why it ran on its runner (oxygenflow#42).
167 
    pub placed: String,
417184c7 Andy McCall ↶ 168 
    /// `waiting` for a person, `approved`, or empty (oxygenflow#57).
169 
    pub approval: String,
170 
    /// Who approved or rejected it, and when.
171 
    pub approved_by: String,
172 
    pub approved_ms: Option<i64>,
05f1a85a Andy McCall ↶ 173 
    /// The agent token it was decided through, if one (oxygenhub#229).
174 
    #[sqlx(default)]
175 
    pub approved_via: String,
b644e175 Andy McCall ↶ 176 
    /// The environment it deploys to, or empty (oxygenflow#59).
177 
    pub environment: String,
47e37402 Andy McCall ↶ 178 
    /// The approval attestation that released it, when sealed (#235).
179 
    pub approval_seal: String,
e97500b7 Andy McCall ↶ 180 
    /// When it entered the queue (made, or run again), and when it became
181 
    /// ready to run: set at its claim (migration 0080, oxygenhub#257).
182 
    #[sqlx(default)]
183 
    pub queued_ms: Option<i64>,
184 
    #[sqlx(default)]
185 
    pub ready_ms: Option<i64>,
ab4df6aa Andy McCall ↶ 186 
    /// When a runner last handed it back, drained or with its lease lapsed
187 
    /// (oxygenflow#76, migration 0094), and how many times: such a job is
188 
    /// queued ahead of new work.
189 
    #[sqlx(default)]
190 
    pub handed_back_ms: Option<i64>,
191 
    #[sqlx(default)]
192 
    pub handbacks: i64,
42fa9554 Andy McCall ↶ 193 
}
194
195 
impl Job {
04683947 Andy McCall ↶ 196 
    /// Where its time went, as its runner reported (oxygenflow#35): each
197 
    /// phase it took, in the order it took them, with its milliseconds.
198 
    pub fn phase_list(&self) -> Vec<(String, u64)> {
199 
        let m: std::collections::BTreeMap<String, u64> = serde_json::from_str(&self.phases).unwrap_or_default();
200 
        ["image", "prepare", "run", "finish"].iter().filter_map(|p| m.get(*p).map(|ms| (p.to_string(), *ms))).collect()
201 
    }
202
42cd46c9 Andy McCall ↶ 203 
    /// The run whose pass this job reused (#265), if it did: in this
204 
    /// repository, or in the one [`Job::reused_repo`] names (#296).
5588bb36 Andy McCall ↶ 205 
    pub fn reused_from(&self) -> Option<i64> {
42cd46c9 Andy McCall ↶ 206 
        self.error.strip_prefix("reused from ")?.rsplit_once("run #")?.1.parse().ok()
5588bb36 Andy McCall ↶ 207 
    }
208
42cd46c9 Andy McCall ↶ 209 
    /// The repository (`owner/name`) whose pass this job reused, when not
210 
    /// its own: a fork reusing what upstream proved (#296).
211 
    pub fn reused_repo(&self) -> Option<&str> {
212 
        let rest = self.error.strip_prefix("reused from ")?;
213 
        let (repo, n) = rest.rsplit_once(" run #")?;
214 
        n.parse::<i64>().ok()?;
215 
        Some(repo).filter(|r| !r.is_empty())
216 
    }
217
42fa9554 Andy McCall ↶ 218 
    pub fn depends_on(&self) -> Vec<String> {
219 
        serde_json::from_str(&self.depends).unwrap_or_default()
220 
    }
221
222 
    pub fn done(&self) -> bool {
745d225d Andy McCall ↶ 223 
        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error" | "cancelled")
42fa9554 Andy McCall ↶ 224 
    }
225 
}
226
227 
#[derive(Clone, Debug, FromRow)]
228 
pub struct LogLine {
229 
    pub seq: i64,
230 
    pub t: i64,
231 
    pub stream: String,
232 
    pub text: String,
233 
}
234
235 
/// A job as the trigger creates it.
236 
pub struct NewJob {
237 
    pub name: String,
238 
    pub depth: i64,
239 
    pub depends: Vec<String>,
240 
    pub spec: String,
241 
    pub sandbox: String,
242 
    pub runtime: String,
44a3d2eb Andy McCall ↶ 243 
    /// The free disk it needs, from [`learned_disk_need`].
244 
    pub need_disk_mib: Option<i64>,
5588bb36 Andy McCall ↶ 245 
    /// A pass on the same state to reuse instead of running it (#265): the
246 
    /// job is created passed, with that pass's receipt.
247 
    pub reused: Option<Reused>,
248 
}
249
250 
/// A job's earlier pass on the same state, spec and sandbox (#265).
251 
#[derive(Clone, Debug)]
252 
pub struct Reused {
42cd46c9 Andy McCall ↶ 253 
    /// The repository it passed in, `owner/name`, when not this one (#296).
254 
    pub repo: Option<String>,
5588bb36 Andy McCall ↶ 255 
    /// The run it passed in.
256 
    pub run_number: i64,
257 
    pub receipt: String,
258 
    pub receipt_hash: String,
259 
    pub receipt_attest: String,
260 
    pub receipt_signer: String,
261 
    pub receipt_note: String,
262 
}
263
264 
impl Reused {
265 
    /// What the reused job says, and how [`Job::reused_from`] reads it back.
266 
    pub fn note(&self) -> String {
42cd46c9 Andy McCall ↶ 267 
        match &self.repo {
268 
            Some(r) => format!("reused from {r} run #{}", self.run_number),
269 
            None => format!("reused from run #{}", self.run_number),
270 
        }
5588bb36 Andy McCall ↶ 271 
    }
272 
}
273
274 
/// The newest pass of job `name` with exactly this spec and sandbox on
275 
/// `state` in this repository (#265).
276 
pub async fn passed_on_state(
277 
    pool: &Pool,
278 
    repo_id: i64,
279 
    state: &str,
280 
    name: &str,
281 
    spec: &str,
282 
    sandbox: &str,
283 
) -> sqlx::Result<Option<Reused>> {
06b75635 Andy McCall ↶ 284 
    passed_in(pool, repo_id, state, name, spec, sandbox, false).await
285 
}
286
287 
/// As [`passed_on_state`]; with `ran_here`, only a job that ran in that
288 
/// repository, not one it reused: its signed receipt is in that
289 
/// repository's store, where another repository's reuse checks it (#296).
290 
async fn passed_in(
291 
    pool: &Pool,
292 
    repo_id: i64,
293 
    state: &str,
294 
    name: &str,
295 
    spec: &str,
296 
    sandbox: &str,
297 
    ran_here: bool,
298 
) -> sqlx::Result<Option<Reused>> {
5588bb36 Andy McCall ↶ 299 
    let row: Option<(i64, String, String, String, String, String)> = sqlx::query_as(
300 
        "SELECT r.number, j.receipt, j.receipt_hash, j.receipt_attest, j.receipt_signer, j.receipt_note \
301 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
302 
         WHERE r.repo_id = $1 AND r.state = $2 AND j.name = $3 AND j.spec = $4 AND j.sandbox = $5 AND j.status = 'passed' \
06b75635 Andy McCall ↶ 303 
           AND (NOT $6 OR j.error NOT LIKE 'reused from %') \
5588bb36 Andy McCall ↶ 304 
         ORDER BY j.id DESC LIMIT 1",
305 
    )
306 
    .bind(repo_id)
307 
    .bind(state)
308 
    .bind(name)
309 
    .bind(spec)
310 
    .bind(sandbox)
06b75635 Andy McCall ↶ 311 
    .bind(ran_here)
5588bb36 Andy McCall ↶ 312 
    .fetch_optional(pool)
313 
    .await?;
314 
    Ok(row.map(|(run_number, receipt, receipt_hash, receipt_attest, receipt_signer, receipt_note)| Reused {
42cd46c9 Andy McCall ↶ 315 
        repo: None,
5588bb36 Andy McCall ↶ 316 
        run_number,
317 
        receipt,
318 
        receipt_hash,
319 
        receipt_attest,
320 
        receipt_signer,
321 
        receipt_note,
322 
    }))
42cd46c9 Andy McCall ↶ 323 
}
324
325 
/// How far up a fork's line [`passed_upstream`] looks.
326 
const UPSTREAM_DEPTH: usize = 5;
327
328 
/// A pass of job `name` with this spec and sandbox on `state` in a
329 
/// repository `repo` was forked from, nearest first (#296): the same state
330 
/// is the same changes, so what upstream proved holds for the fork. Only
331 
/// from an upstream that is public or has the fork's owner, so a private
332 
/// repository's receipts never show in another owner's fork.
333 
pub async fn passed_upstream(
334 
    pool: &Pool,
335 
    repo: &super::repos::RepoRow,
336 
    state: &str,
337 
    name: &str,
338 
    spec: &str,
339 
    sandbox: &str,
340 
) -> sqlx::Result<Option<Reused>> {
341 
    let mut parent = repo.forked_from;
342 
    for _ in 0..UPSTREAM_DEPTH {
343 
        let Some(id) = parent else { break };
344 
        let Some(up) = super::repos::get_by_id(pool, id).await? else { break };
345 
        parent = up.forked_from;
346 
        if !(up.visibility == "public" || up.owner == repo.owner) {
347 
            continue;
348 
        }
06b75635 Andy McCall ↶ 349 
        if let Some(mut r) = passed_in(pool, up.id, state, name, spec, sandbox, true).await? {
42cd46c9 Andy McCall ↶ 350 
            r.repo = Some(format!("{}/{}", up.owner, up.name));
351 
            return Ok(Some(r));
352 
        }
353 
    }
354 
    Ok(None)
42fa9554 Andy McCall ↶ 355 
}
356
91c306aa Andy McCall ↶ 357 
/// A pass of job `name` with this spec and sandbox on `state` in any other
358 
/// repository on the hub (#296): a state is its changes, so whoever proved
359 
/// it, the proof holds here. Only from a repository that is public or has
360 
/// this one's owner, newest first. After [`passed_upstream`], which
361 
/// prefers the repository's own line.
362 
pub async fn passed_elsewhere(
363 
    pool: &Pool,
364 
    repo: &super::repos::RepoRow,
365 
    state: &str,
366 
    name: &str,
367 
    spec: &str,
368 
    sandbox: &str,
369 
) -> sqlx::Result<Option<Reused>> {
370 
    #[derive(FromRow)]
371 
    struct Pass {
372 
        owner: String,
373 
        name: String,
374 
        number: i64,
375 
        receipt: String,
376 
        receipt_hash: String,
377 
        receipt_attest: String,
378 
        receipt_signer: String,
379 
        receipt_note: String,
380 
    }
381 
    let pass: Option<Pass> = sqlx::query_as(
382 
        "SELECT o.owner, o.name, r.number, j.receipt, j.receipt_hash, j.receipt_attest, j.receipt_signer, j.receipt_note \
383 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories o ON o.id = r.repo_id \
384 
         WHERE r.state = $1 AND j.name = $2 AND j.spec = $3 AND j.sandbox = $4 AND j.status = 'passed' \
06b75635 Andy McCall ↶ 385 
           AND o.id <> $5 AND (o.visibility = 'public' OR o.owner = $6) AND j.error NOT LIKE 'reused from %' \
91c306aa Andy McCall ↶ 386 
         ORDER BY j.id DESC LIMIT 1",
387 
    )
388 
    .bind(state)
389 
    .bind(name)
390 
    .bind(spec)
391 
    .bind(sandbox)
392 
    .bind(repo.id)
393 
    .bind(&repo.owner)
394 
    .fetch_optional(pool)
395 
    .await?;
396 
    Ok(pass.map(|p| Reused {
397 
        repo: Some(format!("{}/{}", p.owner, p.name)),
398 
        run_number: p.number,
399 
        receipt: p.receipt,
400 
        receipt_hash: p.receipt_hash,
401 
        receipt_attest: p.receipt_attest,
402 
        receipt_signer: p.receipt_signer,
403 
        receipt_note: p.receipt_note,
404 
    }))
405 
}
406
44a3d2eb Andy McCall ↶ 407 
/// Runs of a job whose peaks set its next need.
408 
const NEED_FROM_RUNS: i64 = 5;
409
410 
/// The free disk a new run of job `name` in repository `repo_id` needs
411 
/// (oxygenflow#41): the most its last runs took, plus a quarter; `None`
412 
/// for a job that has not run here yet.
413 
pub async fn learned_disk_need(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<i64>> {
414 
    let peak: Option<i64> = sqlx::query_scalar(
415 
        "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 \
f0f6c2d9 Andy McCall ↶ 416 
         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",
44a3d2eb Andy McCall ↶ 417 
    )
418 
    .bind(repo_id)
419 
    .bind(name)
420 
    .bind(NEED_FROM_RUNS)
421 
    .fetch_one(pool)
422 
    .await?;
423 
    Ok(peak.map(|p| p + p / 4 + 64))
424 
}
425
501d8b26 Andy McCall ↶ 426 
/// Record why job `job_id` ran where it did.
427 
pub async fn set_placed(pool: &Pool, job_id: i64, placed: &str) -> sqlx::Result<()> {
5d90e0e0 Andy McCall 428 
    sqlx::query("UPDATE flow_jobs SET placed = $1 WHERE id = $2").bind(placed).bind(job_id).execute(pool).await?;
501d8b26 Andy McCall ↶ 429 
    Ok(())
430 
}
431
44a3d2eb Andy McCall ↶ 432 
/// Keep what a job took, as its runner reported.
433 
pub async fn set_peaks(pool: &Pool, job_id: i64, disk_mib: Option<u64>, memory_mib: Option<u64>) -> sqlx::Result<()> {
5d90e0e0 Andy McCall 434 
    sqlx::query("UPDATE flow_jobs SET peak_disk_mib = $1, peak_memory_mib = $2 WHERE id = $3")
44a3d2eb Andy McCall ↶ 435 
        .bind(disk_mib.map(|v| v.min(i64::MAX as u64) as i64))
436 
        .bind(memory_mib.map(|v| v.min(i64::MAX as u64) as i64))
437 
        .bind(job_id)
438 
        .execute(pool)
439 
        .await?;
04683947 Andy McCall ↶ 440 
    Ok(())
441 
}
442
443 
/// Where a finished job's time went, as its runner reported (oxygenflow#35).
444 
pub async fn set_phases(pool: &Pool, job_id: i64, phases: &std::collections::BTreeMap<String, u64>) -> sqlx::Result<()> {
445 
    let json = if phases.is_empty() { String::new() } else { serde_json::to_string(phases).unwrap_or_default() };
446 
    sqlx::query("UPDATE flow_jobs SET phases = $1 WHERE id = $2").bind(json).bind(job_id).execute(pool).await?;
44a3d2eb Andy McCall ↶ 447 
    Ok(())
448 
}
449
42fa9554 Andy McCall ↶ 450 
pub struct NewRun<'a> {
451 
    pub repo_id: i64,
452 
    pub bookmark: &'a str,
453 
    pub state: &'a str,
454 
    pub heads: &'a str,
455 
    pub pipeline_hash: &'a str,
456 
    pub actor: &'a str,
a9f1c0bf Andy McCall ↶ 457 
    /// `push`, `proposal`, `tag:NAME`, `manual`, `schedule` or
458 
    /// `rerun:N:EVENT`.
1edac4b6 Andy McCall ↶ 459 
    pub trigger: &'a str,
42fa9554 Andy McCall ↶ 460 
    /// Problems when the pipeline is invalid: the run is recorded in error.
461 
    pub error: Option<&'a str>,
462 
}
463
367f34b8 Andy McCall ↶ 464 
/// A push that asked for no run (#284): recorded, finished, as skipped,
465 
/// with `why` (which change asked, and who pushed it) as its note.
466 
pub async fn create_skipped_run(pool: &Pool, r: NewRun<'_>, why: &str) -> sqlx::Result<i64> {
467 
    let mut tx = super::begin_write(pool).await?;
468 
    let now = now_ms();
469 
    let id: i64 = sqlx::query_scalar(
470 
        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
471 
         VALUES ($1, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = $2), $3, $4, $5, $6, 'skipped', $7, $8, $9, $10, $11) RETURNING id",
472 
    )
473 
    .bind(r.repo_id)
474 
    .bind(r.repo_id)
475 
    .bind(r.bookmark)
476 
    .bind(r.state)
477 
    .bind(r.heads)
478 
    .bind(r.pipeline_hash)
479 
    .bind(why)
480 
    .bind(r.actor)
481 
    .bind(r.trigger)
482 
    .bind(now)
483 
    .bind(now)
484 
    .fetch_one(&mut *tx)
485 
    .await?;
486 
    tx.commit().await?;
487 
    Ok(id)
488 
}
489
42fa9554 Andy McCall ↶ 490 
pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {
941abdd6 Andy McCall ↶ 491 
    let mut tx = super::begin_write(pool).await?;
42fa9554 Andy McCall ↶ 492 
    let now = now_ms();
493 
    let status = if r.error.is_some() { "error" } else { "queued" };
494 
    let id: i64 = sqlx::query_scalar(
1edac4b6 Andy McCall ↶ 495 
        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
5d90e0e0 Andy McCall 496 
         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",
42fa9554 Andy McCall ↶ 497 
    )
498 
    .bind(r.repo_id)
a0d4bbfb Andy McCall ↶ 499 
    .bind(r.repo_id)
42fa9554 Andy McCall ↶ 500 
    .bind(r.bookmark)
501 
    .bind(r.state)
502 
    .bind(r.heads)
503 
    .bind(r.pipeline_hash)
504 
    .bind(status)
505 
    .bind(r.error.unwrap_or(""))
506 
    .bind(r.actor)
1edac4b6 Andy McCall ↶ 507 
    .bind(r.trigger)
42fa9554 Andy McCall ↶ 508 
    .bind(now)
509 
    .bind(r.error.map(|_| now))
510 
    .fetch_one(&mut *tx)
511 
    .await?;
512 
    for j in jobs {
5588bb36 Andy McCall ↶ 513 
        let job_id: i64 = sqlx::query_scalar(
e97500b7 Andy McCall ↶ 514 
            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib, approval, environment, priority, queued_ms) \
5588bb36 Andy McCall ↶ 515 
             VALUES ($1, $2, $3, $4, $5, $6, $7, 'queued', $8, $9, $10, $11, $12) RETURNING id",
42fa9554 Andy McCall ↶ 516 
        )
517 
        .bind(id)
518 
        .bind(&j.name)
519 
        .bind(j.depth)
520 
        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))
521 
        .bind(&j.spec)
522 
        .bind(&j.sandbox)
523 
        .bind(&j.runtime)
44a3d2eb Andy McCall ↶ 524 
        .bind(j.need_disk_mib)
417184c7 Andy McCall ↶ 525 
        .bind(if needs_approval(&j.spec) { "waiting" } else { "" })
b644e175 Andy McCall ↶ 526 
        .bind(environment_of(&j.spec))
53a0fb2e Andy McCall ↶ 527 
        .bind(priority_of(&j.spec))
e97500b7 Andy McCall ↶ 528 
        .bind(now)
5588bb36 Andy McCall ↶ 529 
        .fetch_one(&mut *tx)
42fa9554 Andy McCall ↶ 530 
        .await?;
5588bb36 Andy McCall ↶ 531 
        // A pass reused (#265): done before any runner can claim it.
532 
        if let Some(r) = &j.reused {
533 
            sqlx::query(
534 
                "UPDATE flow_jobs SET status = 'passed', exit_code = 0, error = $1, receipt = $2, receipt_hash = $3, receipt_attest = $4, \
535 
                 receipt_signer = $5, receipt_note = $6, started_ms = $7, finished_ms = $7 WHERE id = $8",
536 
            )
537 
            .bind(r.note())
538 
            .bind(&r.receipt)
539 
            .bind(&r.receipt_hash)
540 
            .bind(&r.receipt_attest)
541 
            .bind(&r.receipt_signer)
542 
            .bind(&r.receipt_note)
543 
            .bind(now)
544 
            .bind(job_id)
545 
            .execute(&mut *tx)
546 
            .await?;
547 
        }
548 
    }
549 
    // Every job reused: the run has passed already.
550 
    if !jobs.is_empty() && jobs.iter().all(|j| j.reused.is_some()) {
551 
        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = $1 WHERE id = $2").bind(now).bind(id).execute(&mut *tx).await?;
42fa9554 Andy McCall ↶ 552 
    }
553 
    tx.commit().await?;
554 
    Ok(id)
555 
}
556
b644e175 Andy McCall ↶ 557 
/// The environment a compiled job deploys to, or empty (oxygenflow#59).
558 
fn environment_of(spec: &str) -> String {
559 
    serde_json::from_str::<serde_json::Value>(spec).ok().and_then(|v| v["environment"].as_str().map(String::from)).unwrap_or_default()
53a0fb2e Andy McCall ↶ 560 
}
561
562 
/// A compiled job's own priority (`low`, `normal`), or empty; anything
563 
/// else is not one a pipeline may give (oxygenhub#244).
564 
fn priority_of(spec: &str) -> String {
565 
    serde_json::from_str::<serde_json::Value>(spec)
566 
        .ok()
567 
        .and_then(|v| v["priority"].as_str().map(String::from))
568 
        .filter(|p| p == "low" || p == "normal")
569 
        .unwrap_or_default()
b644e175 Andy McCall ↶ 570 
}
571
417184c7 Andy McCall ↶ 572 
/// Whether a compiled job asks for approval (oxygenflow#57).
573 
fn needs_approval(spec: &str) -> bool {
574 
    serde_json::from_str::<serde_json::Value>(spec).is_ok_and(|v| !v["approval"].is_null())
575 
}
576
c4c87f48 Andy McCall ↶ 577 
/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).
578 
fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {
579 
    let mut out: Vec<String> = [spec, sandbox]
580 
        .iter()
581 
        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())
582 
        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())
583 
        .filter_map(|l| l.as_str().map(String::from))
584 
        .collect();
585 
    out.sort();
586 
    out.dedup();
587 
    out
588 
}
589
590 
impl Job {
591 
    /// The runner labels this job needs.
592 
    pub fn needs(&self) -> Vec<String> {
593 
        labels_in(&self.spec, &self.sandbox)
4d2e44d7 Andy McCall ↶ 594 
    }
595
596 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
597 
    pub fn arch(&self) -> Option<String> {
598 
        arch_in(&self.sandbox)
c4c87f48 Andy McCall ↶ 599 
    }
600 
}
601
efd29a00 Andy McCall ↶ 602 
/// The architecture a sandbox's JSON asks for, if any.
603 
fn arch_in(sandbox: &str) -> Option<String> {
604 
    serde_json::from_str::<serde_json::Value>(sandbox).ok()?["arch"].as_str().map(String::from)
605 
}
606
bf3d0754 Andy McCall ↶ 607 
/// A job waiting for or held by a runner, with where it comes from.
608 
#[derive(Clone, Debug, FromRow)]
609 
pub struct QueueEntry {
610 
    pub job_id: i64,
611 
    pub run_id: i64,
612 
    pub repo_id: i64,
613 
    pub owner: String,
614 
    pub name: String,
615 
    pub visibility: String,
616 
    pub run_number: i64,
617 
    pub job: String,
618 
    pub status: String,
b66e70b7 Andy McCall ↶ 619 
    /// The runner holding it, when running.
620 
    pub runner: String,
c4c87f48 Andy McCall ↶ 621 
    pub spec: String,
622 
    pub sandbox: String,
e9b9619c Andy McCall ↶ 623 
    /// When it started running.
624 
    pub started_ms: Option<i64>,
107eaba5 Andy McCall ↶ 625 
    /// Its level (0 urgent … 3 low, oxygenhub#244) and, when expedited, who
626 
    /// did it and why.
627 
    pub priority_rank: i64,
628 
    pub priority_note: String,
c4c87f48 Andy McCall ↶ 629 
}
630
631 
impl QueueEntry {
107eaba5 Andy McCall ↶ 632 
    pub fn priority(&self) -> &'static str {
633 
        PRIORITIES.get(self.priority_rank as usize).copied().unwrap_or("normal")
634 
    }
635 
}
636
637 
impl QueueEntry {
c4c87f48 Andy McCall ↶ 638 
    /// The runner labels the job needs.
639 
    pub fn needs(&self) -> Vec<String> {
640 
        labels_in(&self.spec, &self.sandbox)
efd29a00 Andy McCall ↶ 641 
    }
642
643 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
644 
    pub fn arch(&self) -> Option<String> {
645 
        arch_in(&self.sandbox)
c4c87f48 Andy McCall ↶ 646 
    }
bf3d0754 Andy McCall ↶ 647 
}
648
649 
/// Every job of a live run that is running or queued: running ones first,
650 
/// then queued ones in the order runners claim them (hub spec §13.2).
651 
pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
107eaba5 Andy McCall ↶ 652 
    sqlx::query_as(&format!(
653 
        "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, \
53a0fb2e Andy McCall ↶ 654 
           CAST({PRIORITY_RANK} AS BIGINT) AS priority_rank, {PRIORITY_NOTE} AS priority_note \
107eaba5 Andy McCall ↶ 655 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \
417184c7 Andy McCall ↶ 656 
         WHERE j.status IN ('queued', 'running') AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \
55673c65 Andy McCall ↶ 657 
         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, \
107eaba5 Andy McCall ↶ 658 
           CASE j.status WHEN 'running' THEN 0 ELSE {PRIORITY_RANK} END, \
55673c65 Andy McCall ↶ 659 
           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) \
660 
           + (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, \
107eaba5 Andy McCall ↶ 661 
           j.run_id, j.depth, j.id"
662 
    ))
bf3d0754 Andy McCall ↶ 663 
    .fetch_all(pool)
664 
    .await
665 
}
666
90326c96 Andy McCall ↶ 667 
/// A repository's runs, newest first, numbered below `before`.
668 
pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
5d90e0e0 Andy McCall 669 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND number < $2 ORDER BY number DESC LIMIT $3")
90326c96 Andy McCall ↶ 670 
        .bind(repo_id)
671 
        .bind(before.unwrap_or(i64::MAX))
672 
        .bind(limit)
673 
        .fetch_all(pool)
674 
        .await
675 
}
676
d7ce41ac Andy McCall ↶ 677 
/// The latest run on this bookmark of this repository, for what a push
678 
/// changed since (a job's scope).
745d225d Andy McCall ↶ 679 
/// The run a push to `bookmark` is scoped against: the newest that was not
680 
/// cancelled, nor a push run still going (a new push supersedes it), so
681 
/// that what a cancelled run would have checked is checked by the next.
d7ce41ac Andy McCall ↶ 682 
pub async fn latest_on(pool: &Pool, repo_id: i64, bookmark: &str) -> sqlx::Result<Option<Run>> {
745d225d Andy McCall ↶ 683 
    sqlx::query_as(
367f34b8 Andy McCall ↶ 684 
        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND heads != '' AND status NOT IN ('cancelled', 'skipped')
745d225d Andy McCall ↶ 685 
         AND NOT (trigger = 'push' AND status IN ('queued', 'running')) ORDER BY id DESC LIMIT 1",
686 
    )
687 
    .bind(repo_id)
688 
    .bind(bookmark)
689 
    .fetch_optional(pool)
690 
    .await
d7ce41ac Andy McCall ↶ 691 
}
692
42fa9554 Andy McCall ↶ 693 
pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
5d90e0e0 Andy McCall 694 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND id < $2 ORDER BY id DESC LIMIT $3")
42fa9554 Andy McCall ↶ 695 
        .bind(repo_id)
696 
        .bind(before.unwrap_or(i64::MAX))
697 
        .bind(limit)
698 
        .fetch_all(pool)
699 
        .await
700 
}
701
a0d4bbfb Andy McCall ↶ 702 
/// A repository's run by its number there.
703 
pub async fn run(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<Run>> {
5d90e0e0 Andy McCall 704 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND number = $2").bind(repo_id).bind(number).fetch_optional(pool).await
42fa9554 Andy McCall ↶ 705 
}
706
707 
pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {
5d90e0e0 Andy McCall 708 
    sqlx::query_as("SELECT * FROM flow_runs WHERE id = $1").bind(id).fetch_optional(pool).await
42fa9554 Andy McCall ↶ 709 
}
710
711 
pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {
5d90e0e0 Andy McCall 712 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1 ORDER BY depth, name").bind(run_id).fetch_all(pool).await
42fa9554 Andy McCall ↶ 713 
}
714
715 
pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {
5d90e0e0 Andy McCall 716 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = $1").bind(id).fetch_optional(pool).await
42fa9554 Andy McCall ↶ 717 
}
718
719 
/// The running job a job token belongs to, while it may still be used.
720 
pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {
5d90e0e0 Andy McCall 721 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = $1 AND status = 'running' AND started_ms >= $2")
42fa9554 Andy McCall ↶ 722 
        .bind(token_hash)
723 
        .bind(now_ms() - max_age_ms)
724 
        .fetch_optional(pool)
725 
        .await
726 
}
727
728 
/// Take the oldest queued job whose dependencies all passed and whose tier
729 
/// the runner offers, and mark it running for `runner` with a token.
fd3db644 Andy McCall ↶ 730 
/// Let a job's token also read these repositories (its pinned subrepos).
731 
pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {
732 
    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();
5d90e0e0 Andy McCall 733 
    sqlx::query("UPDATE flow_jobs SET also_read = $1 WHERE id = $2").bind(list.join(",")).bind(job_id).execute(pool).await?;
fd3db644 Andy McCall ↶ 734 
    Ok(())
735 
}
736
737 
impl Job {
738 
    pub fn also_read_ids(&self) -> Vec<i64> {
739 
        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()
740 
    }
741 
}
742
b0612e7d Andy McCall ↶ 743 
/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no
744 
/// limit, and the rootfs images it holds, `None` when it does not say.
745 
#[derive(Clone, Debug, Default)]
76bc5a05 Andy McCall ↶ 746 
pub struct Room {
747 
    pub memory_mib: Option<u32>,
748 
    pub cpus: Option<u32>,
b0612e7d Andy McCall ↶ 749 
    pub images: Option<std::collections::HashSet<String>>,
efd29a00 Andy McCall ↶ 750 
    /// The runner's architecture (`x86_64`, `riscv64`, …), empty when unknown.
751 
    pub arch: String,
752 
    /// Each image's architecture as runners have reported it (`flow_images`).
753 
    pub image_archs: std::collections::HashMap<String, String>,
4d2e44d7 Andy McCall ↶ 754 
    /// Jobs this runner declined, not to be offered to it again (spec §11.5).
755 
    pub declined: std::collections::HashSet<i64>,
92a797b2 Andy McCall ↶ 756 
    /// Images the hub holds that this runner would fetch when it lacks
757 
    /// them (oxygenflow#39); empty for a runner that does not fetch.
758 
    pub fetchable: std::collections::HashSet<String>,
44a3d2eb Andy McCall ↶ 759 
    /// Its free disk now, from its claim's facts (oxygenflow#40); `None`
760 
    /// when it does not say.
761 
    pub free_disk_mib: Option<u64>,
762 
}
763
764 
impl Room {
765 
    /// Whether a job needing `need` MiB of disk fits in its free disk.
766 
    pub fn disk_fits(&self, need: Option<i64>) -> bool {
767 
        match (need, self.free_disk_mib) {
768 
            (Some(n), Some(free)) => n.max(0) as u64 <= free,
769 
            _ => true,
770 
        }
771 
    }
76bc5a05 Andy McCall ↶ 772 
}
773
774 
impl Room {
b0612e7d Andy McCall ↶ 775 
    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,
776 
    /// and its rootfs image among the runner's (oxygenhub#73).
76bc5a05 Andy McCall ↶ 777 
    pub fn fits(&self, sandbox_json: &str) -> bool {
778 
        let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();
779 
        let need = |k: &str| v[k].as_u64().unwrap_or(0);
b0612e7d Andy McCall ↶ 780 
        let rootfs = v["rootfs"].as_str().unwrap_or("");
93501da2 Andy McCall ↶ 781 
        // The architecture: the sandbox's own ask, else the image's as read
782 
        // or learned from its passes (oxygenhub#173); one nobody has read
783 
        // or run yet matches (OxygenFlow spec §11.3).
efd29a00 Andy McCall ↶ 784 
        let wanted = v["arch"].as_str().map(String::from).or_else(|| self.image_archs.get(rootfs).cloned());
785 
        let arch_ok = self.arch.is_empty() || wanted.as_deref().is_none_or(|w| w == self.arch);
786 
        arch_ok
787 
            && self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)
b0612e7d Andy McCall ↶ 788 
            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)
92a797b2 Andy McCall ↶ 789 
            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)) || self.fetchable.contains(rootfs))
efd29a00 Andy McCall ↶ 790 
    }
791 
}
792
793 
/// Remember which architecture each image is for, as a runner reports it.
794 
pub async fn record_image_archs(pool: &Pool, archs: &std::collections::BTreeMap<String, String>) -> sqlx::Result<()> {
795 
    for (hash, arch) in archs.iter().take(500) {
796 
        if hash.len() != 64 || !flow_core::arch_ok(arch) {
797 
            continue;
798 
        }
5d90e0e0 Andy McCall 799 
        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")
efd29a00 Andy McCall ↶ 800 
        .bind(hash.to_ascii_lowercase())
801 
        .bind(arch)
802 
        .bind(now_ms())
803 
        .execute(pool)
804 
        .await?;
76bc5a05 Andy McCall ↶ 805 
    }
efd29a00 Andy McCall ↶ 806 
    Ok(())
807 
}
808
93501da2 Andy McCall ↶ 809 
/// Give every image nobody has read the architecture of the runners its
810 
/// jobs passed on (oxygenhub#173), for the jobs before the hub learned at
811 
/// each pass. Never overrides a known architecture.
812 
pub async fn learn_archs_from_runs(pool: &Pool) -> sqlx::Result<u64> {
3b1bc27f Andy McCall 813 
    let rootfs = crate::db::json_text("j.sandbox", "rootfs");
814 
    let r = sqlx::query(&format!(
93501da2 Andy McCall ↶ 815 
        "INSERT INTO flow_images (hash, arch, seen_ms)
3b1bc27f Andy McCall 816 
         SELECT lower({rootfs}), MIN(r.arch), $1 FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id
817 
         WHERE j.status = 'passed' AND r.arch != '' AND length({rootfs}) = 64
818 
         GROUP BY lower({rootfs})
819 
         ON CONFLICT (hash) DO NOTHING"
820 
    ))
93501da2 Andy McCall ↶ 821 
    .bind(now_ms())
822 
    .execute(pool)
823 
    .await?;
824 
    Ok(r.rows_affected())
825 
}
826
efd29a00 Andy McCall ↶ 827 
/// Every image's architecture the hub knows.
828 
pub async fn image_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
829 
    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, arch FROM flow_images").fetch_all(pool).await?;
830 
    Ok(rows.into_iter().collect())
76bc5a05 Andy McCall ↶ 831 
}
832
42fa9554 Andy McCall ↶ 833 
pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {
c4c87f48 Andy McCall ↶ 834 
    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await
76bc5a05 Andy McCall ↶ 835 
}
836
74be9796 Andy McCall ↶ 837 
/// The first queued job a runner with these tiers, this room and this
55673c65 Andy McCall ↶ 838 
/// scope may take, now running under `runner`. Repositories take turns
839 
/// (fair share, oxygenhub#244): a job's turn is its place in its own
840 
/// repository's queue plus that repository's running jobs, lowest first,
841 
/// oldest first among equals, so one repository's burst can't hold up the
842 
/// rest. A scoped runner never sees
74be9796 Andy McCall ↶ 843 
/// another scope's jobs (oxygenhub#41).
844 
pub async fn claim_within(
845 
    pool: &Pool,
846 
    runtimes: &[String],
847 
    runner: &str,
848 
    token_hash: &str,
849 
    room: Room,
850 
    scope: &super::runners::Scope,
c4c87f48 Andy McCall ↶ 851 
    labels: &[String],
74be9796 Andy McCall ↶ 852 
) -> sqlx::Result<Option<Job>> {
501d8b26 Andy McCall ↶ 853 
    claim_ranked(pool, runtimes, runner, token_hash, room, scope, labels, &|_| false).await
854 
}
855
1ec29691 Andy McCall ↶ 856 
/// Keep the runner build job `id` was claimed by (oxygenflow#84), for
857 
/// counting the jobs that prove a version in a wave.
858 
pub async fn set_runner_version(pool: &Pool, id: i64, version: &str) -> sqlx::Result<()> {
859 
    sqlx::query("UPDATE flow_jobs SET runner_version = $1 WHERE id = $2").bind(version).bind(id).execute(pool).await?;
860 
    Ok(())
861 
}
862
501d8b26 Andy McCall ↶ 863 
/// [`claim_within`], passing over a job `elsewhere` says a better runner
864 
/// should take (oxygenflow#42).
865 
#[allow(clippy::too_many_arguments)]
866 
pub async fn claim_ranked(
867 
    pool: &Pool,
868 
    runtimes: &[String],
869 
    runner: &str,
870 
    token_hash: &str,
871 
    room: Room,
872 
    scope: &super::runners::Scope,
873 
    labels: &[String],
874 
    elsewhere: &(dyn Fn(&Job) -> bool + Sync),
875 
) -> sqlx::Result<Option<Job>> {
941abdd6 Andy McCall ↶ 876 
    let mut tx = super::begin_write(pool).await?;
107eaba5 Andy McCall ↶ 877 
    let queued: Vec<Job> = sqlx::query_as(&format!(
74be9796 Andy McCall ↶ 878 
        "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 \
417184c7 Andy McCall ↶ 879 
         WHERE j.status = 'queued' AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \
5d90e0e0 Andy McCall 880 
         AND ($1 = '' OR rep.owner = $2) AND ($3 IS NULL OR r.repo_id = $4) \
881 
         AND NOT ($5 = '' AND $6 IS NULL AND rep.flow = 'own') \
ab4df6aa Andy McCall ↶ 882 
         ORDER BY CASE WHEN j.handed_back_ms IS NULL THEN 1 ELSE 0 END, {PRIORITY_RANK}, \
883 
           ROW_NUMBER() OVER (PARTITION BY r.repo_id ORDER BY j.run_id, j.depth, j.id) \
55673c65 Andy McCall ↶ 884 
           + (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), \
107eaba5 Andy McCall ↶ 885 
           j.run_id, j.depth, j.id LIMIT 200"
886 
    ))
74be9796 Andy McCall ↶ 887 
    .bind(&scope.owner)
888 
    .bind(&scope.owner)
889 
    .bind(scope.repo_id)
890 
    .bind(scope.repo_id)
1870e4fa Andy McCall ↶ 891 
    // A repository on its own runners only (#225): the site's runners skip it.
892 
    .bind(&scope.owner)
893 
    .bind(scope.repo_id)
42fa9554 Andy McCall ↶ 894 
    .fetch_all(&mut *tx)
895 
    .await?;
896 
    for j in queued {
44a3d2eb Andy McCall ↶ 897 
        if !runtimes.iter().any(|r| r == &j.runtime)
898 
            || !room.fits(&j.sandbox)
899 
            || !room.disk_fits(j.need_disk_mib)
900 
            || room.declined.contains(&j.id)
901 
        {
c4c87f48 Andy McCall ↶ 902 
            continue;
903 
        }
904 
        // Every label the job and its sandbox ask for (`runs-on`).
905 
        if !j.needs().iter().all(|l| labels.contains(l)) {
42fa9554 Andy McCall ↶ 906 
            continue;
907 
        }
908 
        let deps = j.depends_on();
e97500b7 Andy McCall ↶ 909 
        // When it became ready: queued, its dependencies passed, approved.
910 
        let mut ready = j.queued_ms.unwrap_or(0).max(j.approved_ms.unwrap_or(0));
42fa9554 Andy McCall ↶ 911 
        if !deps.is_empty() {
912 
            let mut passed = 0i64;
913 
            for d in &deps {
e97500b7 Andy McCall ↶ 914 
                let st: Option<(String, Option<i64>)> =
915 
                    sqlx::query_as("SELECT status, finished_ms FROM flow_jobs WHERE run_id = $1 AND name = $2")
916 
                        .bind(j.run_id)
917 
                        .bind(d)
918 
                        .fetch_optional(&mut *tx)
919 
                        .await?;
920 
                if let Some((status, finished)) = st {
921 
                    if status == "passed" {
922 
                        passed += 1;
923 
                        ready = ready.max(finished.unwrap_or(0));
924 
                    }
42fa9554 Andy McCall ↶ 925 
                }
926 
            }
927 
            if passed != deps.len() as i64 {
928 
                continue;
929 
            }
501d8b26 Andy McCall ↶ 930 
        }
931 
        if elsewhere(&j) {
932 
            continue;
42fa9554 Andy McCall ↶ 933 
        }
934 
        let now = now_ms();
935 
        let done = sqlx::query(
e97500b7 Andy McCall ↶ 936 
            "UPDATE flow_jobs SET status = 'running', runner = $1, token_hash = $2, started_ms = $3, waiting = '', ready_ms = $5 \
937 
             WHERE id = $4 AND status = 'queued'",
42fa9554 Andy McCall ↶ 938 
        )
939 
        .bind(runner)
940 
        .bind(token_hash)
941 
        .bind(now)
942 
        .bind(j.id)
e97500b7 Andy McCall ↶ 943 
        .bind(if ready > 0 { ready.min(now) } else { now })
42fa9554 Andy McCall ↶ 944 
        .execute(&mut *tx)
945 
        .await?;
946 
        if done.rows_affected() == 1 {
5d90e0e0 Andy McCall 947 
            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = $1 AND status = 'queued'")
42fa9554 Andy McCall ↶ 948 
                .bind(j.run_id)
949 
                .execute(&mut *tx)
950 
                .await?;
951 
            tx.commit().await?;
952 
            return job(pool, j.id).await;
953 
        }
954 
    }
955 
    tx.commit().await?;
956 
    Ok(None)
4d2e44d7 Andy McCall ↶ 957 
}
958
959 
/// How long a decline keeps a job from its runner when the runner's
960 
/// profile stays the same: a reason such as memory in use passes.
961 
pub const DECLINE_HOLDS_MS: i64 = 10 * 60_000;
962
963 
/// The jobs runner `runner_id` declined with this profile, recently.
964 
pub async fn declined_by(pool: &Pool, runner_id: i64, profile: &str) -> sqlx::Result<std::collections::HashSet<i64>> {
5d90e0e0 Andy McCall 965 
    let rows: Vec<(i64,)> = sqlx::query_as("SELECT job_id FROM flow_declines WHERE runner_id = $1 AND profile = $2 AND ms > $3")
4d2e44d7 Andy McCall ↶ 966 
        .bind(runner_id)
967 
        .bind(profile)
968 
        .bind(now_ms() - DECLINE_HOLDS_MS)
969 
        .fetch_all(pool)
970 
        .await?;
971 
    Ok(rows.into_iter().map(|(j,)| j).collect())
972 
}
973
974 
/// A decline: one runner's reason for handing a job back.
975 
#[derive(Clone, Debug, sqlx::FromRow)]
976 
pub struct DeclineRow {
977 
    pub job_id: i64,
978 
    pub runner_id: i64,
979 
    pub runner: String,
980 
    pub reason: String,
981 
    pub ms: i64,
982 
}
983
984 
/// Runner `runner_id` hands back job `job_id`, running on it: the decline
985 
/// is kept (one per job and runner, the latest), and the job is queued
986 
/// again. Returns every decline the job has had, or `None` when the job
987 
/// was not running on that runner.
988 
pub async fn decline(
989 
    pool: &Pool,
990 
    job_id: i64,
991 
    runner_id: i64,
992 
    runner: &str,
993 
    profile: &str,
994 
    reason: &str,
995 
) -> sqlx::Result<Option<Vec<DeclineRow>>> {
996 
    let mut tx = super::begin_write(pool).await?;
997 
    let back = sqlx::query(
998 
        "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, runner_id = NULL \
5d90e0e0 Andy McCall 999 
         WHERE id = $1 AND status = 'running' AND runner_id = $2",
4d2e44d7 Andy McCall ↶ 1000 
    )
1001 
    .bind(job_id)
1002 
    .bind(runner_id)
1003 
    .execute(&mut *tx)
1004 
    .await?;
1005 
    if back.rows_affected() != 1 {
1006 
        tx.rollback().await?;
1007 
        return Ok(None);
1008 
    }
1009 
    sqlx::query(
5d90e0e0 Andy McCall 1010 
        "INSERT INTO flow_declines (job_id, runner_id, runner, profile, reason, ms) VALUES ($1, $2, $3, $4, $5, $6) \
4d2e44d7 Andy McCall ↶ 1011 
         ON CONFLICT (job_id, runner_id) DO UPDATE SET runner = excluded.runner, profile = excluded.profile, reason = excluded.reason, ms = excluded.ms",
1012 
    )
1013 
    .bind(job_id)
1014 
    .bind(runner_id)
1015 
    .bind(runner)
1016 
    .bind(profile)
1017 
    .bind(reason)
1018 
    .bind(now_ms())
1019 
    .execute(&mut *tx)
1020 
    .await?;
1021 
    let all: Vec<DeclineRow> =
5d90e0e0 Andy McCall 1022 
        sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = $1 ORDER BY ms")
4d2e44d7 Andy McCall ↶ 1023 
            .bind(job_id)
1024 
            .fetch_all(&mut *tx)
1025 
            .await?;
1026 
    tx.commit().await?;
1027 
    Ok(Some(all))
ab4df6aa Andy McCall ↶ 1028 
}
1029
1030 
/// Job `job_id`, running on runner `runner_id` (any runner when `None`),
1031 
/// goes back to the queue ahead of new work (oxygenflow#76): drained, or
1032 
/// its runner's lease lapsed. Its log is kept and its next attempt's
1033 
/// follows on. Returns how many times it has now been handed back, or
1034 
/// `None` when it was not running there.
1035 
pub async fn hand_back(pool: &Pool, job_id: i64, runner_id: Option<i64>) -> sqlx::Result<Option<i64>> {
1036 
    let mut tx = super::begin_write(pool).await?;
1037 
    let now = now_ms();
1038 
    let back = sqlx::query(
1039 
        "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, runner_id = NULL, \
1040 
         handed_back_ms = $1, handbacks = handbacks + 1, queued_ms = $2, ready_ms = NULL, placed = '' \
1041 
         WHERE id = $3 AND status = 'running' AND ($4 IS NULL OR runner_id = $5)",
1042 
    )
1043 
    .bind(now)
1044 
    .bind(now)
1045 
    .bind(job_id)
1046 
    .bind(runner_id)
1047 
    .bind(runner_id)
1048 
    .execute(&mut *tx)
1049 
    .await?;
1050 
    if back.rows_affected() != 1 {
1051 
        tx.rollback().await?;
1052 
        return Ok(None);
1053 
    }
1054 
    let n: i64 = sqlx::query_scalar("SELECT handbacks FROM flow_jobs WHERE id = $1").bind(job_id).fetch_one(&mut *tx).await?;
1055 
    tx.commit().await?;
1056 
    Ok(Some(n))
1057 
}
1058
3b650652 Andy McCall ↶ 1059 
/// How long job `name` of repository `repo_id` is expected to take: the
1060 
/// median of its last five passing runs (claimed to finished), passes
1061 
/// reused from another run left out (oxygenflow#76). `None` before it has
1062 
/// passed: a runner leaving then hands it back rather than guess.
1063 
pub async fn expected_ms(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<i64>> {
1064 
    let mut times: Vec<i64> = sqlx::query_scalar(
1065 
        "SELECT j.finished_ms - j.started_ms FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
1066 
         WHERE r.repo_id = $1 AND j.name = $2 AND j.status = 'passed' AND j.error = '' \
1067 
         AND j.started_ms IS NOT NULL AND j.finished_ms IS NOT NULL ORDER BY j.id DESC LIMIT 5",
1068 
    )
1069 
    .bind(repo_id)
1070 
    .bind(name)
1071 
    .fetch_all(pool)
1072 
    .await?;
1073 
    times.sort_unstable();
1074 
    Ok(times.get(times.len() / 2).copied().filter(|ms| *ms > 0))
1075 
}
1076
ab4df6aa Andy McCall ↶ 1077 
/// Jobs running on a runner the hub has not heard from since `before`
1078 
/// (claims, heartbeats, logs): their leases lapsed (oxygenflow#76).
1079 
pub async fn running_on_silent_runners(pool: &Pool, before: i64) -> sqlx::Result<Vec<(Job, String)>> {
1080 
    let rows: Vec<(i64, String)> = sqlx::query_as(
1081 
        "SELECT j.id, r.name FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id \
1082 
         WHERE j.status = 'running' AND r.last_seen_ms < $1 ORDER BY j.id",
1083 
    )
1084 
    .bind(before)
1085 
    .fetch_all(pool)
1086 
    .await?;
1087 
    let mut out = Vec::new();
1088 
    for (id, runner) in rows {
1089 
        if let Some(j) = job(pool, id).await? {
1090 
            out.push((j, runner));
1091 
        }
1092 
    }
1093 
    Ok(out)
4d2e44d7 Andy McCall ↶ 1094 
}
1095
1096 
/// The declines of job `job_id`, oldest first.
1097 
pub async fn declines_of_job(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<DeclineRow>> {
5d90e0e0 Andy McCall 1098 
    sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = $1 ORDER BY ms")
4d2e44d7 Andy McCall ↶ 1099 
        .bind(job_id)
1100 
        .fetch_all(pool)
1101 
        .await
1102 
}
1103
1104 
/// A runner's latest declines, newest first, with where the job is.
1105 
#[derive(Clone, Debug, sqlx::FromRow)]
1106 
pub struct RunnerDecline {
1107 
    pub job_id: i64,
1108 
    pub job: String,
1109 
    pub run_number: i64,
1110 
    pub owner: String,
1111 
    pub name: String,
1112 
    pub reason: String,
1113 
    pub ms: i64,
1114 
}
1115
1116 
pub async fn declines_of_runner(pool: &Pool, runner_id: i64, limit: i64) -> sqlx::Result<Vec<RunnerDecline>> {
1117 
    sqlx::query_as(
1118 
        "SELECT d.job_id, j.name AS job, r.number AS run_number, rep.owner, rep.name, d.reason, d.ms \
1119 
         FROM flow_declines d JOIN flow_jobs j ON j.id = d.job_id JOIN flow_runs r ON r.id = j.run_id \
5d90e0e0 Andy McCall 1120 
         JOIN repositories rep ON rep.id = r.repo_id WHERE d.runner_id = $1 ORDER BY d.ms DESC LIMIT $2",
4d2e44d7 Andy McCall ↶ 1121 
    )
1122 
    .bind(runner_id)
1123 
    .bind(limit)
1124 
    .fetch_all(pool)
1125 
    .await
1126 
}
1127
1128 
/// Queued jobs whose dependencies have passed, in their live runs, with
1129 
/// their repositories (owner, id): the ones a runner could take now.
58cbf424 Andy McCall ↶ 1130 
/// Every queued job's priority rank, 0 (urgent) to 3 (low), by job id
1131 
/// (#259: how many ready jobs come first).
1132 
pub async fn queued_ranks(pool: &Pool) -> sqlx::Result<std::collections::HashMap<i64, i64>> {
1133 
    let rows: Vec<(i64, i64)> = sqlx::query_as(&format!(
1134 
        "SELECT j.id, CAST({PRIORITY_RANK} AS BIGINT) FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \
1135 
         WHERE j.status = 'queued'"
1136 
    ))
1137 
    .fetch_all(pool)
1138 
    .await?;
1139 
    Ok(rows.into_iter().collect())
1140 
}
1141
1142 
/// How many jobs each runner is running now, by runner id (#259).
1143 
pub async fn running_by_runner(pool: &Pool) -> sqlx::Result<std::collections::HashMap<i64, i64>> {
1144 
    let rows: Vec<(i64, i64)> =
1145 
        sqlx::query_as("SELECT runner_id, COUNT(*) FROM flow_jobs WHERE status = 'running' AND runner_id IS NOT NULL GROUP BY runner_id")
1146 
            .fetch_all(pool)
1147 
            .await?;
1148 
    Ok(rows.into_iter().collect())
1149 
}
1150
4d2e44d7 Andy McCall ↶ 1151 
pub async fn ready_queued(pool: &Pool) -> sqlx::Result<Vec<(Job, String, i64)>> {
1152 
    let queued: Vec<Job> = sqlx::query_as(
1153 
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
417184c7 Andy McCall ↶ 1154 
         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",
4d2e44d7 Andy McCall ↶ 1155 
    )
1156 
    .fetch_all(pool)
1157 
    .await?;
1158 
    let mut out = Vec::new();
1159 
    for j in queued {
1160 
        let mut ready = true;
1161 
        for d in j.depends_on() {
5d90e0e0 Andy McCall 1162 
            let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = $1 AND name = $2")
4d2e44d7 Andy McCall ↶ 1163 
                .bind(j.run_id)
1164 
                .bind(&d)
1165 
                .fetch_optional(pool)
1166 
                .await?;
1167 
            if st.as_deref() != Some("passed") {
1168 
                ready = false;
1169 
                break;
1170 
            }
1171 
        }
1172 
        if !ready {
1173 
            continue;
1174 
        }
1175 
        let repo: Option<(String, i64)> =
5d90e0e0 Andy McCall 1176 
            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")
4d2e44d7 Andy McCall ↶ 1177 
                .bind(j.run_id)
1178 
                .fetch_optional(pool)
1179 
                .await?;
1180 
        if let Some((owner, id)) = repo {
1181 
            out.push((j, owner, id));
1182 
        }
1183 
    }
1184 
    Ok(out)
1185 
}
1186
1187 
pub async fn set_waiting(pool: &Pool, job_id: i64, waiting: &str) -> sqlx::Result<()> {
5d90e0e0 Andy McCall 1188 
    sqlx::query("UPDATE flow_jobs SET waiting = $1 WHERE id = $2 AND waiting <> $3")
4d2e44d7 Andy McCall ↶ 1189 
        .bind(waiting)
1190 
        .bind(job_id)
1191 
        .bind(waiting)
1192 
        .execute(pool)
1193 
        .await?;
1194 
    Ok(())
42fa9554 Andy McCall ↶ 1195 
}
1196
1197 
/// Append records; returns how many were kept (the rest exceed the cap).
1198 
pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {
941abdd6 Andy McCall ↶ 1199 
    let mut tx = super::begin_write(pool).await?;
5d90e0e0 Andy McCall 1200 
    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = $1").bind(job_id).fetch_one(&mut *tx).await?;
42fa9554 Andy McCall ↶ 1201 
    let room = (cap - count).max(0) as usize;
1202 
    let keep = &records[..records.len().min(room)];
1203 
    for (i, r) in keep.iter().enumerate() {
5d90e0e0 Andy McCall 1204 
        sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES ($1, $2, $3, $4, $5)")
42fa9554 Andy McCall ↶ 1205 
            .bind(job_id)
1206 
            .bind(count + i as i64 + 1)
1207 
            .bind(r.t as i64)
1208 
            .bind(&r.stream)
1209 
            .bind(&r.text)
1210 
            .execute(&mut *tx)
1211 
            .await?;
1212 
    }
5d90e0e0 Andy McCall 1213 
    sqlx::query("UPDATE flow_jobs SET log_count = log_count + $1 WHERE id = $2")
42fa9554 Andy McCall ↶ 1214 
        .bind(keep.len() as i64)
1215 
        .bind(job_id)
1216 
        .execute(&mut *tx)
1217 
        .await?;
1218 
    tx.commit().await?;
1219 
    Ok(keep.len())
1220 
}
1221
1222 
pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {
5d90e0e0 Andy McCall 1223 
    sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = $1 AND seq > $2 ORDER BY seq LIMIT $3")
42fa9554 Andy McCall ↶ 1224 
        .bind(job_id)
1225 
        .bind(after)
1226 
        .bind(limit)
1227 
        .fetch_all(pool)
1228 
        .await
1229 
}
1230
070a7b64 Andy McCall ↶ 1231 
/// Jobs running since before `ms` (the oldest first).
5dead1f4 Andy McCall ↶ 1232 
/// Jobs still running at `now` past their own time limit (`default_s` for
1233 
/// one claimed without) plus `grace_ms`.
1234 
pub async fn running_past_limit(pool: &Pool, now: i64, default_s: u64, grace_ms: i64) -> sqlx::Result<Vec<Job>> {
1235 
    sqlx::query_as(
1236 
        "SELECT * FROM flow_jobs WHERE status = 'running' \
5d90e0e0 Andy McCall 1237 
         AND COALESCE(started_ms, 0) + (CASE WHEN timeout_s > 0 THEN timeout_s ELSE $1 END) * 1000 + $2 < $3 ORDER BY started_ms",
5dead1f4 Andy McCall ↶ 1238 
    )
1239 
    .bind(default_s as i64)
1240 
    .bind(grace_ms)
1241 
    .bind(now)
1242 
    .fetch_all(pool)
1243 
    .await
1244 
}
1245
1246 
/// Record the limit a claimed job runs under.
1247 
pub async fn set_job_timeout(pool: &Pool, job_id: i64, timeout_s: u64) -> sqlx::Result<()> {
5d90e0e0 Andy McCall 1248 
    sqlx::query("UPDATE flow_jobs SET timeout_s = $1 WHERE id = $2").bind(timeout_s as i64).bind(job_id).execute(pool).await?;
5dead1f4 Andy McCall ↶ 1249 
    Ok(())
1250 
}
1251
070a7b64 Andy McCall ↶ 1252 
pub async fn running_since_before(pool: &Pool, ms: i64) -> sqlx::Result<Vec<Job>> {
5d90e0e0 Andy McCall 1253 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < $1 ORDER BY started_ms")
070a7b64 Andy McCall ↶ 1254 
        .bind(ms)
1255 
        .fetch_all(pool)
1256 
        .await
1257 
}
1258
eef677c4 Andy McCall ↶ 1259 
/// The jobs running on the runner with this id.
1260 
pub async fn running_on(pool: &Pool, runner_id: i64) -> sqlx::Result<Vec<Job>> {
5d90e0e0 Andy McCall 1261 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = $1 ORDER BY id").bind(runner_id).fetch_all(pool).await
eef677c4 Andy McCall ↶ 1262 
}
1263
42fa9554 Andy McCall ↶ 1264 
pub struct Outcome<'a> {
1265 
    pub status: &'a str,
1266 
    pub exit_code: Option<i64>,
1267 
    pub error: &'a str,
1268 
    pub receipt: &'a str,
1269 
    pub receipt_hash: &'a str,
1270 
}
1271
1272 
/// Record a job's end, skip the jobs that can no longer run, and settle the
1273 
/// run. Returns the jobs whose status changed (for their checks).
1274 
pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {
941abdd6 Andy McCall ↶ 1275 
    let mut tx = super::begin_write(pool).await?;
42fa9554 Andy McCall ↶ 1276 
    let now = now_ms();
5d90e0e0 Andy McCall 1277 
    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = $1").bind(job_id).fetch_one(&mut *tx).await?;
1278 
    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")
42fa9554 Andy McCall ↶ 1279 
        .bind(o.status)
1280 
        .bind(o.exit_code)
1281 
        .bind(o.error)
1282 
        .bind(o.receipt)
1283 
        .bind(o.receipt_hash)
1284 
        .bind(now)
1285 
        .bind(job_id)
1286 
        .execute(&mut *tx)
1287 
        .await?;
93501da2 Andy McCall ↶ 1288 
    // A pass teaches what its image is for, when nobody has read it
1289 
    // (oxygenhub#173): the architecture of the runner it passed on.
1290 
    if o.status == "passed" {
1291 
        let rootfs = serde_json::from_str::<serde_json::Value>(&j.sandbox)
1292 
            .ok()
1293 
            .and_then(|v| v["rootfs"].as_str().map(str::to_ascii_lowercase))
1294 
            .unwrap_or_default();
1295 
        if rootfs.len() == 64 {
1296 
            sqlx::query(
5d90e0e0 Andy McCall 1297 
                "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 \
1298 
                 WHERE j.id = $3 AND r.arch != '' ON CONFLICT (hash) DO NOTHING",
93501da2 Andy McCall ↶ 1299 
            )
1300 
            .bind(&rootfs)
1301 
            .bind(now)
1302 
            .bind(job_id)
1303 
            .execute(&mut *tx)
1304 
            .await?;
1305 
        }
1306 
    }
42fa9554 Andy McCall ↶ 1307 
    let mut changed = vec![job_id];
1308 
    // Dependents of anything that did not pass are skipped, transitively.
1309 
    loop {
5d90e0e0 Andy McCall 1310 
        let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1").bind(j.run_id).fetch_all(&mut *tx).await?;
42fa9554 Andy McCall ↶ 1311 
        let blocked: Vec<i64> = all
1312 
            .iter()
1313 
            .filter(|x| x.status == "queued")
1314 
            .filter(|x| {
1315 
                x.depends_on()
1316 
                    .iter()
417184c7 Andy McCall ↶ 1317 
                    .any(|d| all.iter().any(|y| &y.name == d && matches!(y.status.as_str(), "failed" | "skipped" | "error" | "cancelled")))
42fa9554 Andy McCall ↶ 1318 
            })
1319 
            .map(|x| x.id)
1320 
            .collect();
1321 
        if blocked.is_empty() {
1322 
            break;
1323 
        }
1324 
        for id in blocked {
5d90e0e0 Andy McCall 1325 
            sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = $1 WHERE id = $2")
42fa9554 Andy McCall ↶ 1326 
                .bind(now)
1327 
                .bind(id)
1328 
                .execute(&mut *tx)
1329 
                .await?;
1330 
            changed.push(id);
1331 
        }
1332 
    }
5d90e0e0 Andy McCall 1333 
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1").bind(j.run_id).fetch_all(&mut *tx).await?;
42fa9554 Andy McCall ↶ 1334 
    if all.iter().all(Job::done) {
1335 
        let status = if all.iter().all(|x| x.status == "passed") { "passed" } else { "failed" };
5d90e0e0 Andy McCall 1336 
        sqlx::query("UPDATE flow_runs SET status = $1, finished_ms = $2 WHERE id = $3")
42fa9554 Andy McCall ↶ 1337 
            .bind(status)
1338 
            .bind(now)
1339 
            .bind(j.run_id)
1340 
            .execute(&mut *tx)
1341 
            .await?;
1342 
    }
1343 
    tx.commit().await?;
1344 
    let mut out = Vec::new();
1345 
    for id in changed {
745d225d Andy McCall ↶ 1346 
        if let Some(x) = job(pool, id).await? {
1347 
            out.push(x);
1348 
        }
1349 
    }
1350 
    Ok(out)
1351 
}
1352
1353 
/// Cancel a run (oxygenflow#32): its jobs not yet done end as cancelled,
1354 
/// and so does the run. A runner learns of it when the hub next refuses
1355 
/// its log lines, and stops the job. Returns the jobs whose status changed.
1356 
pub async fn cancel_run(pool: &Pool, run_id: i64, why: &str) -> sqlx::Result<Vec<Job>> {
1357 
    let now = now_ms();
1358 
    let mut tx = pool.begin().await?;
5d90e0e0 Andy McCall 1359 
    let ids: Vec<(i64,)> = sqlx::query_as("SELECT id FROM flow_jobs WHERE run_id = $1 AND status IN ('queued', 'running')")
745d225d Andy McCall ↶ 1360 
        .bind(run_id)
1361 
        .fetch_all(&mut *tx)
1362 
        .await?;
1363 
    sqlx::query(
5d90e0e0 Andy McCall 1364 
        "UPDATE flow_jobs SET status = 'cancelled', error = $1, finished_ms = $2, token_hash = ''
1365 
         WHERE run_id = $3 AND status IN ('queued', 'running')",
745d225d Andy McCall ↶ 1366 
    )
1367 
    .bind(why)
1368 
    .bind(now)
1369 
    .bind(run_id)
1370 
    .execute(&mut *tx)
1371 
    .await?;
5d90e0e0 Andy McCall 1372 
    sqlx::query("UPDATE flow_runs SET status = 'cancelled', finished_ms = $1 WHERE id = $2 AND status IN ('queued', 'running')")
745d225d Andy McCall ↶ 1373 
        .bind(now)
1374 
        .bind(run_id)
1375 
        .execute(&mut *tx)
1376 
        .await?;
1377 
    tx.commit().await?;
1378 
    let mut out = Vec::new();
1379 
    for (id,) in ids {
42fa9554 Andy McCall ↶ 1380 
        if let Some(x) = job(pool, id).await? {
1381 
            out.push(x);
1382 
        }
1383 
    }
1384 
    Ok(out)
745d225d Andy McCall ↶ 1385 
}
1386
e8fa5369 Andy McCall ↶ 1387 
/// Run job `name` of finished run `run_id` again, with every job that
1388 
/// depends on it (oxygenflow#31): back to the queue as if new, their logs,
1389 
/// artifacts and reports gone, and the run running again. The jobs they
1390 
/// depend on keep their results. The jobs reset; none when there is no
1391 
/// such job.
1392 
pub async fn rerun_job(pool: &Pool, run_id: i64, name: &str) -> sqlx::Result<Vec<Job>> {
5d90e0e0 Andy McCall 1393 
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = $1").bind(run_id).fetch_all(pool).await?;
e8fa5369 Andy McCall ↶ 1394 
    if !all.iter().any(|j| j.name == name) {
1395 
        return Ok(Vec::new());
1396 
    }
1397 
    let mut again = vec![name.to_string()];
1398 
    loop {
1399 
        let more: Vec<String> = all
1400 
            .iter()
1401 
            .filter(|j| !again.contains(&j.name) && j.depends_on().iter().any(|d| again.contains(d)))
1402 
            .map(|j| j.name.clone())
1403 
            .collect();
1404 
        if more.is_empty() {
1405 
            break;
1406 
        }
1407 
        again.extend(more);
1408 
    }
1409 
    let ids: Vec<i64> = all.iter().filter(|j| again.contains(&j.name)).map(|j| j.id).collect();
1410 
    let mut tx = pool.begin().await?;
1411 
    for id in &ids {
1412 
        for table in ["flow_logs", "flow_artifacts", "flow_declines", "test_reports", "coverage_reports", "reports"] {
5d90e0e0 Andy McCall 1413 
            sqlx::query(&format!("DELETE FROM {table} WHERE job_id = $1")).bind(id).execute(&mut *tx).await?;
e8fa5369 Andy McCall ↶ 1414 
        }
1415 
        sqlx::query(
1416 
            "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, finished_ms = NULL,
1417 
             exit_code = NULL, error = '', receipt = '', receipt_hash = '', log_count = 0, also_read = '', receipt_attest = '',
1418 
             receipt_signer = '', receipt_note = '', runner_id = NULL, waiting = '', placed = '', peak_disk_mib = NULL,
04683947 Andy McCall ↶ 1419 
             peak_memory_mib = NULL, phases = '', approval = CASE WHEN approval = '' THEN '' ELSE 'waiting' END, approved_by = '',
ab4df6aa Andy McCall ↶ 1420 
             approved_ms = NULL, approval_seal = '', approved_via = '', queued_ms = $2, ready_ms = NULL, handed_back_ms = NULL,
1421 
             handbacks = 0 WHERE id = $1",
e8fa5369 Andy McCall ↶ 1422 
        )
1423 
        .bind(id)
e97500b7 Andy McCall ↶ 1424 
        .bind(now_ms())
e8fa5369 Andy McCall ↶ 1425 
        .execute(&mut *tx)
1426 
        .await?;
1427 
    }
5d90e0e0 Andy McCall 1428 
    sqlx::query("UPDATE flow_runs SET status = 'running', finished_ms = NULL WHERE id = $1").bind(run_id).execute(&mut *tx).await?;
e8fa5369 Andy McCall ↶ 1429 
    tx.commit().await?;
1430 
    let mut out = Vec::new();
1431 
    for id in ids {
1432 
        if let Some(x) = job(pool, id).await? {
1433 
            out.push(x);
1434 
        }
1435 
    }
1436 
    Ok(out)
1437 
}
1438
a9f1c0bf Andy McCall ↶ 1439 
/// The runs of `bookmark` started by `trigger` still going, older than run
1440 
/// `number`: those a newer run supersedes (#267). With `running_too`
1441 
/// false, only those still queued.
1442 
pub async fn live_runs_before(
1443 
    pool: &Pool,
1444 
    repo_id: i64,
1445 
    bookmark: &str,
1446 
    trigger: &str,
1447 
    number: i64,
1448 
    running_too: bool,
1449 
) -> sqlx::Result<Vec<Run>> {
745d225d Andy McCall ↶ 1450 
    sqlx::query_as(
a9f1c0bf Andy McCall ↶ 1451 
        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND number < $3 AND trigger = $4
1452 
         AND (status = 'queued' OR ($5 AND status = 'running')) ORDER BY id",
745d225d Andy McCall ↶ 1453 
    )
1454 
    .bind(repo_id)
1455 
    .bind(bookmark)
1456 
    .bind(number)
a9f1c0bf Andy McCall ↶ 1457 
    .bind(trigger)
1458 
    .bind(running_too)
745d225d Andy McCall ↶ 1459 
    .fetch_all(pool)
1460 
    .await
42fa9554 Andy McCall ↶ 1461 
}
736f6813 Andy McCall ↶ 1462
2abea308 Andy McCall ↶ 1463 
/// The runs on `bookmark` that no runner has started yet: queued, with
1464 
/// the given trigger, created before `before_ms` (#263).
1465 
pub async fn unstarted_runs_on(pool: &Pool, repo_id: i64, bookmark: &str, trigger: &str, before_ms: i64) -> sqlx::Result<Vec<Run>> {
1466 
    sqlx::query_as(
1467 
        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND trigger = $3 AND status = 'queued' AND created_ms <= $4 ORDER BY id",
1468 
    )
1469 
    .bind(repo_id)
1470 
    .bind(bookmark)
1471 
    .bind(trigger)
1472 
    .bind(before_ms)
1473 
    .fetch_all(pool)
1474 
    .await
1475 
}
1476
736f6813 Andy McCall ↶ 1477 
/// The newest run of exactly this state (heads joined with `+`, sorted).
1478 
pub async fn latest_for_heads(pool: &Pool, repo_id: i64, heads: &str) -> sqlx::Result<Option<Run>> {
5d90e0e0 Andy McCall 1479 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND heads = $2 ORDER BY id DESC LIMIT 1")
736f6813 Andy McCall ↶ 1480 
        .bind(repo_id)
1481 
        .bind(heads)
1482 
        .fetch_optional(pool)
1483 
        .await
cc19697f Andy McCall ↶ 1484 
}
1485
1486 
/// Record what became of job `job_id`'s signed receipt.
1487 
pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {
5d90e0e0 Andy McCall 1488 
    sqlx::query("UPDATE flow_jobs SET receipt_attest = $1, receipt_signer = $2, receipt_note = $3 WHERE id = $4")
cc19697f Andy McCall ↶ 1489 
        .bind(attest)
1490 
        .bind(signer)
1491 
        .bind(note)
1492 
        .bind(job_id)
1493 
        .execute(pool)
1494 
        .await?;
1495 
    Ok(())
736f6813 Andy McCall ↶ 1496 
}
a3c8288b Andy McCall ↶ 1497
1498 
/// A job's artifact, kept as a blob of the run's repository.
1499 
#[derive(Clone, Debug, FromRow)]
1500 
pub struct Artifact {
1501 
    pub job_id: i64,
1502 
    pub name: String,
1503 
    pub path: String,
1504 
    pub blob: String,
1505 
    pub size: i64,
1506 
    /// BLAKE3 of the bytes.
1507 
    pub content: String,
1508 
    pub created_ms: i64,
1509 
}
1510
1511 
pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {
5d90e0e0 Andy McCall 1512 
    sqlx::query(
d1bf2d69 Andy McCall ↶ 1513 
        "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",
5d90e0e0 Andy McCall 1514 
    )
1515 
    .bind(a.job_id)
1516 
    .bind(&a.name)
1517 
    .bind(&a.path)
1518 
    .bind(&a.blob)
1519 
    .bind(a.size)
1520 
    .bind(&a.content)
1521 
    .bind(a.created_ms)
1522 
    .execute(pool)
1523 
    .await?;
a3c8288b Andy McCall ↶ 1524 
    Ok(())
1525 
}
1526
45977501 Andy McCall ↶ 1527 
/// Every artifact blob a repository's runs keep, for garbage collection to
1528 
/// keep too (#126): the database is the only thing that names them.
1529 
pub async fn artifact_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
1530 
    sqlx::query_scalar(
5d90e0e0 Andy McCall 1531 
        "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",
45977501 Andy McCall ↶ 1532 
    )
1533 
    .bind(repo_id)
1534 
    .fetch_all(pool)
1535 
    .await
1536 
}
1537
e4203e14 Andy McCall ↶ 1538 
/// Job `job_id`'s artifacts, by name; not its [`CHECKPOINT`], which no
1539 
/// receipt names.
a3c8288b Andy McCall ↶ 1540 
pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
e4203e14 Andy McCall ↶ 1541 
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = $1 AND name NOT LIKE '.%' ORDER BY name").bind(job_id).fetch_all(pool).await
a3c8288b Andy McCall ↶ 1542 
}
10708ab8 Andy McCall ↶ 1543
e4203e14 Andy McCall ↶ 1544 
/// The name a job's checkpoint is kept under (oxygenflow#77), beside its
1545 
/// artifacts: no pipeline can declare it.
1546 
pub const CHECKPOINT: &str = ".checkpoint";
1547
1548 
/// The checkpoint an attempt of job `job_id` saved as its runner handed it
1549 
/// back (oxygenflow#77), if one did.
1550 
pub async fn checkpoint(pool: &Pool, job_id: i64) -> sqlx::Result<Option<Artifact>> {
1551 
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = $1 AND name = $2").bind(job_id).bind(CHECKPOINT).fetch_optional(pool).await
1552 
}
1553
16ee5d19 Andy McCall ↶ 1554 
/// An artifact with the run and job that made it, for a repository's list
1555 
/// of every run's outputs (oxygenhub#254).
1556 
#[derive(Clone, Debug, FromRow)]
1557 
pub struct RepoArtifact {
1558 
    pub run_number: i64,
1559 
    pub bookmark: String,
1560 
    pub job: String,
1561 
    pub name: String,
1562 
    pub path: String,
1563 
    pub size: i64,
1564 
    pub content: String,
1565 
    pub created_ms: i64,
1566 
}
1567
1568 
/// A repository's newest artifacts, at most `limit`.
1569 
pub async fn repo_artifacts(pool: &Pool, repo_id: i64, limit: i64) -> sqlx::Result<Vec<RepoArtifact>> {
1570 
    sqlx::query_as(
1571 
        "SELECT r.number AS run_number, r.bookmark, j.name AS job, a.name, a.path, a.size, a.content, a.created_ms
1572 
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id
1573 
         WHERE r.repo_id = $1 ORDER BY a.created_ms DESC, r.number DESC, j.name, a.name LIMIT $2",
1574 
    )
1575 
    .bind(repo_id)
1576 
    .bind(limit)
1577 
    .fetch_all(pool)
1578 
    .await
1579 
}
1580
10708ab8 Andy McCall ↶ 1581 
/// Where an artifact with given content came from: the run and job that
1582 
/// built it, and the job's signed receipt when one was kept.
1583 
#[derive(Clone, Debug, FromRow)]
1584 
pub struct Provenance {
1585 
    pub run_number: i64,
1586 
    pub job: String,
1587 
    pub name: String,
1588 
    pub receipt_attest: String,
1589 
    pub receipt_signer: String,
1590 
}
1591
1592 
/// The build of the bytes whose BLAKE3 is `content` in repository
1593 
/// `repo_id`: a passed job, one with a kept signed receipt first, newest
1594 
/// run first.
1595 
pub async fn provenance(pool: &Pool, repo_id: i64, content: &str) -> sqlx::Result<Option<Provenance>> {
1596 
    sqlx::query_as(
1597 
        "SELECT r.number AS run_number, j.name AS job, a.name AS name, j.receipt_attest, j.receipt_signer \
1598 
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id \
5d90e0e0 Andy McCall 1599 
         WHERE r.repo_id = $1 AND a.content = $2 AND j.status = 'passed' \
10708ab8 Andy McCall ↶ 1600 
         ORDER BY (j.receipt_attest != '') DESC, r.number DESC LIMIT 1",
1601 
    )
1602 
    .bind(repo_id)
1603 
    .bind(content)
1604 
    .fetch_optional(pool)
417184c7 Andy McCall ↶ 1605 
    .await
1606 
}
1607
1608 
/// Approve job `id` if it is still waiting for a person (oxygenflow#57);
1609 
/// whether it was.
05f1a85a Andy McCall ↶ 1610 
pub async fn approve_job(pool: &Pool, id: i64, who: &str, seal: &str, via: &str) -> sqlx::Result<bool> {
417184c7 Andy McCall ↶ 1611 
    let r = sqlx::query(
05f1a85a Andy McCall ↶ 1612 
        "UPDATE flow_jobs SET approval = 'approved', approved_by = $1, approved_ms = $2, approval_seal = $3, approved_via = $5 \
5d90e0e0 Andy McCall 1613 
         WHERE id = $4 AND status = 'queued' AND approval = 'waiting'",
417184c7 Andy McCall ↶ 1614 
    )
1615 
    .bind(who)
1616 
    .bind(now_ms())
47e37402 Andy McCall ↶ 1617 
    .bind(seal)
417184c7 Andy McCall ↶ 1618 
    .bind(id)
05f1a85a Andy McCall ↶ 1619 
    .bind(via)
417184c7 Andy McCall ↶ 1620 
    .execute(pool)
1621 
    .await?;
1622 
    Ok(r.rows_affected() > 0)
1623 
}
1624
1625 
/// Note who rejected job `id` (it is then finished as cancelled).
05f1a85a Andy McCall ↶ 1626 
pub async fn note_rejection(pool: &Pool, id: i64, who: &str, via: &str) -> sqlx::Result<()> {
1627 
    sqlx::query("UPDATE flow_jobs SET approved_by = $1, approved_ms = $2, approved_via = $4 WHERE id = $3")
417184c7 Andy McCall ↶ 1628 
        .bind(who)
1629 
        .bind(now_ms())
1630 
        .bind(id)
05f1a85a Andy McCall ↶ 1631 
        .bind(via)
417184c7 Andy McCall ↶ 1632 
        .execute(pool)
1633 
        .await?;
1634 
    Ok(())
1635 
}
1636
1637 
/// Jobs waiting for a person, in runs still going.
1638 
pub async fn waiting_for_approval(pool: &Pool) -> sqlx::Result<Vec<Job>> {
1639 
    sqlx::query_as(
1640 
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
1641 
         WHERE j.status = 'queued' AND j.approval = 'waiting' AND r.status IN ('queued', 'running')",
1642 
    )
b644e175 Andy McCall ↶ 1643 
    .fetch_all(pool)
1644 
    .await
1645 
}
1646
1647 
/// One deployment: a job for an environment that passed.
1648 
#[derive(Clone, Debug, FromRow)]
1649 
pub struct Deployment {
1650 
    pub environment: String,
1651 
    pub job: String,
1652 
    pub run_number: i64,
1653 
    pub state: String,
1654 
    pub bookmark: String,
1655 
    pub actor: String,
1656 
    pub approved_by: String,
1657 
    pub finished_ms: i64,
1658 
}
1659
1660 
/// A repository's deployments, newest first (oxygenflow#59).
1661 
pub async fn deployments(pool: &Pool, repo_id: i64, limit: i64) -> sqlx::Result<Vec<Deployment>> {
1662 
    sqlx::query_as(
1663 
        "SELECT j.environment, j.name AS job, r.number AS run_number, r.state, r.bookmark, r.actor, j.approved_by, \
1664 
         COALESCE(j.finished_ms, 0) AS finished_ms FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
5d90e0e0 Andy McCall 1665 
         WHERE r.repo_id = $1 AND j.environment != '' AND j.status = 'passed' ORDER BY j.finished_ms DESC LIMIT $2",
b644e175 Andy McCall ↶ 1666 
    )
1667 
    .bind(repo_id)
1668 
    .bind(limit)
417184c7 Andy McCall ↶ 1669 
    .fetch_all(pool)
10708ab8 Andy McCall ↶ 1670 
    .await
1671 
}