OXYGENHUB
# 2fb70d019851 oxygenhub / crates / hub / src / db / flow.rs
31.8 KiB · Rust · last changed Images by hash on the hub: chunks, verified manifests, and placement for runners that fetch 24 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 
}
26
27 
impl Run {
28 
    /// How the run page says what started it.
29 
    pub fn started_by(&self) -> String {
30 
        match self.trigger.as_str() {
31 
            "manual" => format!("started by {}", self.actor),
32 
            "schedule" => "started on schedule".to_string(),
33 
            t if t.starts_with("tag:") => format!("tag {} pushed by {}", &t[4..], self.actor),
34 
            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", self.rerun_of().unwrap_or(""), self.actor),
35 
            _ => format!("pushed by {}", self.actor),
36 
        }
37 
    }
38
39 
    /// The run a rerun repeats: `N` of `rerun:N` or `rerun:N:EVENT`.
40 
    pub fn rerun_of(&self) -> Option<&str> {
41 
        let rest = self.trigger.strip_prefix("rerun:")?;
42 
        Some(rest.split(':').next().unwrap_or(rest))
43 
    }
44
45 
    /// The trigger as people read it: `rerun:N`, without the event it repeats.
46 
    pub fn trigger_label(&self) -> String {
47 
        match self.rerun_of() {
48 
            Some(n) => format!("rerun:{n}"),
49 
            None => self.trigger.clone(),
50 
        }
51 
    }
52
53 
    pub fn head_list(&self) -> Vec<&str> {
54 
        self.heads.split('+').filter(|h| !h.is_empty()).collect()
55 
    }
56
57 
    pub fn live(&self) -> bool {
58 
        matches!(self.status.as_str(), "queued" | "running")
59 
    }
60 
}
61
62 
#[derive(Clone, Debug, Default, FromRow)]
63 
pub struct Job {
64 
    pub id: i64,
65 
    pub run_id: i64,
66 
    pub name: String,
67 
    pub depth: i64,
68 
    pub depends: String,
69 
    pub spec: String,
70 
    pub sandbox: String,
71 
    pub runtime: String,
72 
    pub status: String,
73 
    pub runner: String,
74 
    pub token_hash: String,
75 
    pub started_ms: Option<i64>,
76 
    pub finished_ms: Option<i64>,
77 
    pub exit_code: Option<i64>,
78 
    pub error: String,
79 
    pub receipt: String,
80 
    pub receipt_hash: String,
81 
    pub log_count: i64,
82 
    /// Comma-separated ids of other repositories the job token may read.
83 
    pub also_read: String,
84 
    /// The receipt attestation kept in the repository (Oxygen spec §14.6).
85 
    pub receipt_attest: String,
86 
    /// Who signed it: the runner's identity name.
87 
    pub receipt_signer: String,
88 
    /// Why a signed receipt was not kept.
89 
    pub receipt_note: String,
90 
    /// Why a queued job no runner can take waits (spec §11.4), from the
91 
    /// last placement pass; empty when it is only waiting its turn.
92 
    pub waiting: String,
93 
}
94
95 
impl Job {
96 
    pub fn depends_on(&self) -> Vec<String> {
97 
        serde_json::from_str(&self.depends).unwrap_or_default()
98 
    }
99
100 
    pub fn done(&self) -> bool {
101 
        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error" | "cancelled")
102 
    }
103 
}
104
105 
#[derive(Clone, Debug, FromRow)]
106 
pub struct LogLine {
107 
    pub seq: i64,
108 
    pub t: i64,
109 
    pub stream: String,
110 
    pub text: String,
111 
}
112
113 
/// A job as the trigger creates it.
114 
pub struct NewJob {
115 
    pub name: String,
116 
    pub depth: i64,
117 
    pub depends: Vec<String>,
118 
    pub spec: String,
119 
    pub sandbox: String,
120 
    pub runtime: String,
121 
}
122
123 
pub struct NewRun<'a> {
124 
    pub repo_id: i64,
125 
    pub bookmark: &'a str,
126 
    pub state: &'a str,
127 
    pub heads: &'a str,
128 
    pub pipeline_hash: &'a str,
129 
    pub actor: &'a str,
130 
    /// `push`, `manual` or `rerun:N`.
131 
    pub trigger: &'a str,
132 
    /// Problems when the pipeline is invalid: the run is recorded in error.
133 
    pub error: Option<&'a str>,
134 
}
135
136 
pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {
137 
    let mut tx = super::begin_write(pool).await?;
138 
    let now = now_ms();
139 
    let status = if r.error.is_some() { "error" } else { "queued" };
140 
    let id: i64 = sqlx::query_scalar(
141 
        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
142 
         VALUES (?, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
143 
    )
144 
    .bind(r.repo_id)
145 
    .bind(r.repo_id)
146 
    .bind(r.bookmark)
147 
    .bind(r.state)
148 
    .bind(r.heads)
149 
    .bind(r.pipeline_hash)
150 
    .bind(status)
151 
    .bind(r.error.unwrap_or(""))
152 
    .bind(r.actor)
153 
    .bind(r.trigger)
154 
    .bind(now)
155 
    .bind(r.error.map(|_| now))
156 
    .fetch_one(&mut *tx)
157 
    .await?;
158 
    for j in jobs {
159 
        sqlx::query(
160 
            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued')",
161 
        )
162 
        .bind(id)
163 
        .bind(&j.name)
164 
        .bind(j.depth)
165 
        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))
166 
        .bind(&j.spec)
167 
        .bind(&j.sandbox)
168 
        .bind(&j.runtime)
169 
        .execute(&mut *tx)
170 
        .await?;
171 
    }
172 
    tx.commit().await?;
173 
    Ok(id)
174 
}
175
176 
/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).
177 
fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {
178 
    let mut out: Vec<String> = [spec, sandbox]
179 
        .iter()
180 
        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())
181 
        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())
182 
        .filter_map(|l| l.as_str().map(String::from))
183 
        .collect();
184 
    out.sort();
185 
    out.dedup();
186 
    out
187 
}
188
189 
impl Job {
190 
    /// The runner labels this job needs.
191 
    pub fn needs(&self) -> Vec<String> {
192 
        labels_in(&self.spec, &self.sandbox)
193 
    }
194
195 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
196 
    pub fn arch(&self) -> Option<String> {
197 
        arch_in(&self.sandbox)
198 
    }
199 
}
200
201 
/// The architecture a sandbox's JSON asks for, if any.
202 
fn arch_in(sandbox: &str) -> Option<String> {
203 
    serde_json::from_str::<serde_json::Value>(sandbox).ok()?["arch"].as_str().map(String::from)
204 
}
205
206 
/// A job waiting for or held by a runner, with where it comes from.
207 
#[derive(Clone, Debug, FromRow)]
208 
pub struct QueueEntry {
209 
    pub job_id: i64,
210 
    pub run_id: i64,
211 
    pub repo_id: i64,
212 
    pub owner: String,
213 
    pub name: String,
214 
    pub visibility: String,
215 
    pub run_number: i64,
216 
    pub job: String,
217 
    pub status: String,
218 
    /// The runner holding it, when running.
219 
    pub runner: String,
220 
    pub spec: String,
221 
    pub sandbox: String,
222 
}
223
224 
impl QueueEntry {
225 
    /// The runner labels the job needs.
226 
    pub fn needs(&self) -> Vec<String> {
227 
        labels_in(&self.spec, &self.sandbox)
228 
    }
229
230 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
231 
    pub fn arch(&self) -> Option<String> {
232 
        arch_in(&self.sandbox)
233 
    }
234 
}
235
236 
/// Every job of a live run that is running or queued: running ones first,
237 
/// then queued ones in the order runners claim them (hub spec §13.2).
238 
pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
239 
    sqlx::query_as(
240 
        "SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox \
241 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
242 
         WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \
243 
         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, j.run_id, j.depth, j.id",
244 
    )
245 
    .fetch_all(pool)
246 
    .await
247 
}
248
249 
/// A repository's runs, newest first, numbered below `before`.
250 
pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
251 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number < ? ORDER BY number DESC LIMIT ?")
252 
        .bind(repo_id)
253 
        .bind(before.unwrap_or(i64::MAX))
254 
        .bind(limit)
255 
        .fetch_all(pool)
256 
        .await
257 
}
258
259 
/// The latest run on this bookmark of this repository, for what a push
260 
/// changed since (a job's scope).
261 
/// The run a push to `bookmark` is scoped against: the newest that was not
262 
/// cancelled, nor a push run still going (a new push supersedes it), so
263 
/// that what a cancelled run would have checked is checked by the next.
264 
pub async fn latest_on(pool: &Pool, repo_id: i64, bookmark: &str) -> sqlx::Result<Option<Run>> {
265 
    sqlx::query_as(
266 
        "SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND heads != '' AND status != 'cancelled'
267 
         AND NOT (trigger = 'push' AND status IN ('queued', 'running')) ORDER BY id DESC LIMIT 1",
268 
    )
269 
    .bind(repo_id)
270 
    .bind(bookmark)
271 
    .fetch_optional(pool)
272 
    .await
273 
}
274
275 
pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
276 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND id < ? ORDER BY id DESC LIMIT ?")
277 
        .bind(repo_id)
278 
        .bind(before.unwrap_or(i64::MAX))
279 
        .bind(limit)
280 
        .fetch_all(pool)
281 
        .await
282 
}
283
284 
/// A repository's run by its number there.
285 
pub async fn run(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<Run>> {
286 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number = ?").bind(repo_id).bind(number).fetch_optional(pool).await
287 
}
288
289 
pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {
290 
    sqlx::query_as("SELECT * FROM flow_runs WHERE id = ?").bind(id).fetch_optional(pool).await
291 
}
292
293 
pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {
294 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ? ORDER BY depth, name").bind(run_id).fetch_all(pool).await
295 
}
296
297 
pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {
298 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(id).fetch_optional(pool).await
299 
}
300
301 
/// The running job a job token belongs to, while it may still be used.
302 
pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {
303 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = ? AND status = 'running' AND started_ms >= ?")
304 
        .bind(token_hash)
305 
        .bind(now_ms() - max_age_ms)
306 
        .fetch_optional(pool)
307 
        .await
308 
}
309
310 
/// Take the oldest queued job whose dependencies all passed and whose tier
311 
/// the runner offers, and mark it running for `runner` with a token.
312 
/// Let a job's token also read these repositories (its pinned subrepos).
313 
pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {
314 
    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();
315 
    sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;
316 
    Ok(())
317 
}
318
319 
impl Job {
320 
    pub fn also_read_ids(&self) -> Vec<i64> {
321 
        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()
322 
    }
323 
}
324
325 
/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no
326 
/// limit, and the rootfs images it holds, `None` when it does not say.
327 
#[derive(Clone, Debug, Default)]
328 
pub struct Room {
329 
    pub memory_mib: Option<u32>,
330 
    pub cpus: Option<u32>,
331 
    pub images: Option<std::collections::HashSet<String>>,
332 
    /// The runner's architecture (`x86_64`, `riscv64`, …), empty when unknown.
333 
    pub arch: String,
334 
    /// Each image's architecture as runners have reported it (`flow_images`).
335 
    pub image_archs: std::collections::HashMap<String, String>,
336 
    /// Jobs this runner declined, not to be offered to it again (spec §11.5).
337 
    pub declined: std::collections::HashSet<i64>,
338 
    /// Images the hub holds that this runner would fetch when it lacks
339 
    /// them (oxygenflow#39); empty for a runner that does not fetch.
340 
    pub fetchable: std::collections::HashSet<String>,
341 
}
342
343 
impl Room {
344 
    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,
345 
    /// and its rootfs image among the runner's (oxygenhub#73).
346 
    pub fn fits(&self, sandbox_json: &str) -> bool {
347 
        let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();
348 
        let need = |k: &str| v[k].as_u64().unwrap_or(0);
349 
        let rootfs = v["rootfs"].as_str().unwrap_or("");
350 
        // The architecture: the sandbox's own ask, else the image's as known;
351 
        // an unknown one matches, as before (OxygenFlow spec §11.3).
352 
        let wanted = v["arch"].as_str().map(String::from).or_else(|| self.image_archs.get(rootfs).cloned());
353 
        let arch_ok = self.arch.is_empty() || wanted.as_deref().is_none_or(|w| w == self.arch);
354 
        arch_ok
355 
            && self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)
356 
            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)
357 
            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)) || self.fetchable.contains(rootfs))
358 
    }
359 
}
360
361 
/// Remember which architecture each image is for, as a runner reports it.
362 
pub async fn record_image_archs(pool: &Pool, archs: &std::collections::BTreeMap<String, String>) -> sqlx::Result<()> {
363 
    for (hash, arch) in archs.iter().take(500) {
364 
        if hash.len() != 64 || !flow_core::arch_ok(arch) {
365 
            continue;
366 
        }
367 
        sqlx::query("INSERT INTO flow_images (hash, arch, seen_ms) VALUES (?, ?, ?) ON CONFLICT (hash) DO UPDATE SET arch = excluded.arch, seen_ms = excluded.seen_ms")
368 
        .bind(hash.to_ascii_lowercase())
369 
        .bind(arch)
370 
        .bind(now_ms())
371 
        .execute(pool)
372 
        .await?;
373 
    }
374 
    Ok(())
375 
}
376
377 
/// Every image's architecture the hub knows.
378 
pub async fn image_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
379 
    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, arch FROM flow_images").fetch_all(pool).await?;
380 
    Ok(rows.into_iter().collect())
381 
}
382
383 
pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {
384 
    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await
385 
}
386
387 
/// The oldest runnable job that fits in `room`.
388 
/// The first queued job a runner with these tiers, this room and this
389 
/// scope may take, now running under `runner`. A scoped runner never sees
390 
/// another scope's jobs (oxygenhub#41).
391 
pub async fn claim_within(
392 
    pool: &Pool,
393 
    runtimes: &[String],
394 
    runner: &str,
395 
    token_hash: &str,
396 
    room: Room,
397 
    scope: &super::runners::Scope,
398 
    labels: &[String],
399 
) -> sqlx::Result<Option<Job>> {
400 
    let mut tx = super::begin_write(pool).await?;
401 
    let queued: Vec<Job> = sqlx::query_as(
402 
        "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 \
403 
         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') \
404 
         AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \
405 
         ORDER BY j.run_id, j.depth, j.id LIMIT 200",
406 
    )
407 
    .bind(&scope.owner)
408 
    .bind(&scope.owner)
409 
    .bind(scope.repo_id)
410 
    .bind(scope.repo_id)
411 
    .fetch_all(&mut *tx)
412 
    .await?;
413 
    for j in queued {
414 
        if !runtimes.iter().any(|r| r == &j.runtime) || !room.fits(&j.sandbox) || room.declined.contains(&j.id) {
415 
            continue;
416 
        }
417 
        // Every label the job and its sandbox ask for (`runs-on`).
418 
        if !j.needs().iter().all(|l| labels.contains(l)) {
419 
            continue;
420 
        }
421 
        let deps = j.depends_on();
422 
        if !deps.is_empty() {
423 
            let mut passed = 0i64;
424 
            for d in &deps {
425 
                let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")
426 
                    .bind(j.run_id)
427 
                    .bind(d)
428 
                    .fetch_optional(&mut *tx)
429 
                    .await?;
430 
                if st.as_deref() == Some("passed") {
431 
                    passed += 1;
432 
                }
433 
            }
434 
            if passed != deps.len() as i64 {
435 
                continue;
436 
            }
437 
        }
438 
        let now = now_ms();
439 
        let done = sqlx::query(
440 
            "UPDATE flow_jobs SET status = 'running', runner = ?, token_hash = ?, started_ms = ? WHERE id = ? AND status = 'queued'",
441 
        )
442 
        .bind(runner)
443 
        .bind(token_hash)
444 
        .bind(now)
445 
        .bind(j.id)
446 
        .execute(&mut *tx)
447 
        .await?;
448 
        if done.rows_affected() == 1 {
449 
            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = ? AND status = 'queued'")
450 
                .bind(j.run_id)
451 
                .execute(&mut *tx)
452 
                .await?;
453 
            tx.commit().await?;
454 
            return job(pool, j.id).await;
455 
        }
456 
    }
457 
    tx.commit().await?;
458 
    Ok(None)
459 
}
460
461 
/// How long a decline keeps a job from its runner when the runner's
462 
/// profile stays the same: a reason such as memory in use passes.
463 
pub const DECLINE_HOLDS_MS: i64 = 10 * 60_000;
464
465 
/// The jobs runner `runner_id` declined with this profile, recently.
466 
pub async fn declined_by(pool: &Pool, runner_id: i64, profile: &str) -> sqlx::Result<std::collections::HashSet<i64>> {
467 
    let rows: Vec<(i64,)> = sqlx::query_as("SELECT job_id FROM flow_declines WHERE runner_id = ? AND profile = ? AND ms > ?")
468 
        .bind(runner_id)
469 
        .bind(profile)
470 
        .bind(now_ms() - DECLINE_HOLDS_MS)
471 
        .fetch_all(pool)
472 
        .await?;
473 
    Ok(rows.into_iter().map(|(j,)| j).collect())
474 
}
475
476 
/// A decline: one runner's reason for handing a job back.
477 
#[derive(Clone, Debug, sqlx::FromRow)]
478 
pub struct DeclineRow {
479 
    pub job_id: i64,
480 
    pub runner_id: i64,
481 
    pub runner: String,
482 
    pub reason: String,
483 
    pub ms: i64,
484 
}
485
486 
/// Runner `runner_id` hands back job `job_id`, running on it: the decline
487 
/// is kept (one per job and runner, the latest), and the job is queued
488 
/// again. Returns every decline the job has had, or `None` when the job
489 
/// was not running on that runner.
490 
pub async fn decline(
491 
    pool: &Pool,
492 
    job_id: i64,
493 
    runner_id: i64,
494 
    runner: &str,
495 
    profile: &str,
496 
    reason: &str,
497 
) -> sqlx::Result<Option<Vec<DeclineRow>>> {
498 
    let mut tx = super::begin_write(pool).await?;
499 
    let back = sqlx::query(
500 
        "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, runner_id = NULL \
501 
         WHERE id = ? AND status = 'running' AND runner_id = ?",
502 
    )
503 
    .bind(job_id)
504 
    .bind(runner_id)
505 
    .execute(&mut *tx)
506 
    .await?;
507 
    if back.rows_affected() != 1 {
508 
        tx.rollback().await?;
509 
        return Ok(None);
510 
    }
511 
    sqlx::query(
512 
        "INSERT INTO flow_declines (job_id, runner_id, runner, profile, reason, ms) VALUES (?, ?, ?, ?, ?, ?) \
513 
         ON CONFLICT (job_id, runner_id) DO UPDATE SET runner = excluded.runner, profile = excluded.profile, reason = excluded.reason, ms = excluded.ms",
514 
    )
515 
    .bind(job_id)
516 
    .bind(runner_id)
517 
    .bind(runner)
518 
    .bind(profile)
519 
    .bind(reason)
520 
    .bind(now_ms())
521 
    .execute(&mut *tx)
522 
    .await?;
523 
    let all: Vec<DeclineRow> =
524 
        sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = ? ORDER BY ms")
525 
            .bind(job_id)
526 
            .fetch_all(&mut *tx)
527 
            .await?;
528 
    tx.commit().await?;
529 
    Ok(Some(all))
530 
}
531
532 
/// The declines of job `job_id`, oldest first.
533 
pub async fn declines_of_job(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<DeclineRow>> {
534 
    sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = ? ORDER BY ms")
535 
        .bind(job_id)
536 
        .fetch_all(pool)
537 
        .await
538 
}
539
540 
/// A runner's latest declines, newest first, with where the job is.
541 
#[derive(Clone, Debug, sqlx::FromRow)]
542 
pub struct RunnerDecline {
543 
    pub job_id: i64,
544 
    pub job: String,
545 
    pub run_number: i64,
546 
    pub owner: String,
547 
    pub name: String,
548 
    pub reason: String,
549 
    pub ms: i64,
550 
}
551
552 
pub async fn declines_of_runner(pool: &Pool, runner_id: i64, limit: i64) -> sqlx::Result<Vec<RunnerDecline>> {
553 
    sqlx::query_as(
554 
        "SELECT d.job_id, j.name AS job, r.number AS run_number, rep.owner, rep.name, d.reason, d.ms \
555 
         FROM flow_declines d JOIN flow_jobs j ON j.id = d.job_id JOIN flow_runs r ON r.id = j.run_id \
556 
         JOIN repositories rep ON rep.id = r.repo_id WHERE d.runner_id = ? ORDER BY d.ms DESC LIMIT ?",
557 
    )
558 
    .bind(runner_id)
559 
    .bind(limit)
560 
    .fetch_all(pool)
561 
    .await
562 
}
563
564 
/// Queued jobs whose dependencies have passed, in their live runs, with
565 
/// their repositories (owner, id): the ones a runner could take now.
566 
pub async fn ready_queued(pool: &Pool) -> sqlx::Result<Vec<(Job, String, i64)>> {
567 
    let queued: Vec<Job> = sqlx::query_as(
568 
        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \
569 
         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') ORDER BY j.run_id, j.depth, j.id LIMIT 500",
570 
    )
571 
    .fetch_all(pool)
572 
    .await?;
573 
    let mut out = Vec::new();
574 
    for j in queued {
575 
        let mut ready = true;
576 
        for d in j.depends_on() {
577 
            let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")
578 
                .bind(j.run_id)
579 
                .bind(&d)
580 
                .fetch_optional(pool)
581 
                .await?;
582 
            if st.as_deref() != Some("passed") {
583 
                ready = false;
584 
                break;
585 
            }
586 
        }
587 
        if !ready {
588 
            continue;
589 
        }
590 
        let repo: Option<(String, i64)> =
591 
            sqlx::query_as("SELECT rep.owner, rep.id FROM flow_runs r JOIN repositories rep ON rep.id = r.repo_id WHERE r.id = ?")
592 
                .bind(j.run_id)
593 
                .fetch_optional(pool)
594 
                .await?;
595 
        if let Some((owner, id)) = repo {
596 
            out.push((j, owner, id));
597 
        }
598 
    }
599 
    Ok(out)
600 
}
601
602 
pub async fn set_waiting(pool: &Pool, job_id: i64, waiting: &str) -> sqlx::Result<()> {
603 
    sqlx::query("UPDATE flow_jobs SET waiting = ? WHERE id = ? AND waiting <> ?")
604 
        .bind(waiting)
605 
        .bind(job_id)
606 
        .bind(waiting)
607 
        .execute(pool)
608 
        .await?;
609 
    Ok(())
610 
}
611
612 
/// Append records; returns how many were kept (the rest exceed the cap).
613 
pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {
614 
    let mut tx = super::begin_write(pool).await?;
615 
    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
616 
    let room = (cap - count).max(0) as usize;
617 
    let keep = &records[..records.len().min(room)];
618 
    for (i, r) in keep.iter().enumerate() {
619 
        sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES (?, ?, ?, ?, ?)")
620 
            .bind(job_id)
621 
            .bind(count + i as i64 + 1)
622 
            .bind(r.t as i64)
623 
            .bind(&r.stream)
624 
            .bind(&r.text)
625 
            .execute(&mut *tx)
626 
            .await?;
627 
    }
628 
    sqlx::query("UPDATE flow_jobs SET log_count = log_count + ? WHERE id = ?")
629 
        .bind(keep.len() as i64)
630 
        .bind(job_id)
631 
        .execute(&mut *tx)
632 
        .await?;
633 
    tx.commit().await?;
634 
    Ok(keep.len())
635 
}
636
637 
pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {
638 
    sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")
639 
        .bind(job_id)
640 
        .bind(after)
641 
        .bind(limit)
642 
        .fetch_all(pool)
643 
        .await
644 
}
645
646 
/// Jobs running since before `ms` (the oldest first).
647 
/// Jobs still running at `now` past their own time limit (`default_s` for
648 
/// one claimed without) plus `grace_ms`.
649 
pub async fn running_past_limit(pool: &Pool, now: i64, default_s: u64, grace_ms: i64) -> sqlx::Result<Vec<Job>> {
650 
    sqlx::query_as(
651 
        "SELECT * FROM flow_jobs WHERE status = 'running' \
652 
         AND COALESCE(started_ms, 0) + (CASE WHEN timeout_s > 0 THEN timeout_s ELSE ? END) * 1000 + ? < ? ORDER BY started_ms",
653 
    )
654 
    .bind(default_s as i64)
655 
    .bind(grace_ms)
656 
    .bind(now)
657 
    .fetch_all(pool)
658 
    .await
659 
}
660
661 
/// Record the limit a claimed job runs under.
662 
pub async fn set_job_timeout(pool: &Pool, job_id: i64, timeout_s: u64) -> sqlx::Result<()> {
663 
    sqlx::query("UPDATE flow_jobs SET timeout_s = ? WHERE id = ?").bind(timeout_s as i64).bind(job_id).execute(pool).await?;
664 
    Ok(())
665 
}
666
667 
pub async fn running_since_before(pool: &Pool, ms: i64) -> sqlx::Result<Vec<Job>> {
668 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < ? ORDER BY started_ms")
669 
        .bind(ms)
670 
        .fetch_all(pool)
671 
        .await
672 
}
673
674 
/// The jobs running on the runner with this id.
675 
pub async fn running_on(pool: &Pool, runner_id: i64) -> sqlx::Result<Vec<Job>> {
676 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = ? ORDER BY id").bind(runner_id).fetch_all(pool).await
677 
}
678
679 
pub struct Outcome<'a> {
680 
    pub status: &'a str,
681 
    pub exit_code: Option<i64>,
682 
    pub error: &'a str,
683 
    pub receipt: &'a str,
684 
    pub receipt_hash: &'a str,
685 
}
686
687 
/// Record a job's end, skip the jobs that can no longer run, and settle the
688 
/// run. Returns the jobs whose status changed (for their checks).
689 
pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {
690 
    let mut tx = super::begin_write(pool).await?;
691 
    let now = now_ms();
692 
    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
693 
    sqlx::query("UPDATE flow_jobs SET status = ?, exit_code = ?, error = ?, receipt = ?, receipt_hash = ?, finished_ms = ?, token_hash = '' WHERE id = ?")
694 
        .bind(o.status)
695 
        .bind(o.exit_code)
696 
        .bind(o.error)
697 
        .bind(o.receipt)
698 
        .bind(o.receipt_hash)
699 
        .bind(now)
700 
        .bind(job_id)
701 
        .execute(&mut *tx)
702 
        .await?;
703 
    let mut changed = vec![job_id];
704 
    // Dependents of anything that did not pass are skipped, transitively.
705 
    loop {
706 
        let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
707 
        let blocked: Vec<i64> = all
708 
            .iter()
709 
            .filter(|x| x.status == "queued")
710 
            .filter(|x| {
711 
                x.depends_on()
712 
                    .iter()
713 
                    .any(|d| all.iter().any(|y| &y.name == d && matches!(y.status.as_str(), "failed" | "skipped" | "error")))
714 
            })
715 
            .map(|x| x.id)
716 
            .collect();
717 
        if blocked.is_empty() {
718 
            break;
719 
        }
720 
        for id in blocked {
721 
            sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = ? WHERE id = ?")
722 
                .bind(now)
723 
                .bind(id)
724 
                .execute(&mut *tx)
725 
                .await?;
726 
            changed.push(id);
727 
        }
728 
    }
729 
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
730 
    if all.iter().all(Job::done) {
731 
        let status = if all.iter().all(|x| x.status == "passed") { "passed" } else { "failed" };
732 
        sqlx::query("UPDATE flow_runs SET status = ?, finished_ms = ? WHERE id = ?")
733 
            .bind(status)
734 
            .bind(now)
735 
            .bind(j.run_id)
736 
            .execute(&mut *tx)
737 
            .await?;
738 
    }
739 
    tx.commit().await?;
740 
    let mut out = Vec::new();
741 
    for id in changed {
742 
        if let Some(x) = job(pool, id).await? {
743 
            out.push(x);
744 
        }
745 
    }
746 
    Ok(out)
747 
}
748
749 
/// Cancel a run (oxygenflow#32): its jobs not yet done end as cancelled,
750 
/// and so does the run. A runner learns of it when the hub next refuses
751 
/// its log lines, and stops the job. Returns the jobs whose status changed.
752 
pub async fn cancel_run(pool: &Pool, run_id: i64, why: &str) -> sqlx::Result<Vec<Job>> {
753 
    let now = now_ms();
754 
    let mut tx = pool.begin().await?;
755 
    let ids: Vec<(i64,)> = sqlx::query_as("SELECT id FROM flow_jobs WHERE run_id = ? AND status IN ('queued', 'running')")
756 
        .bind(run_id)
757 
        .fetch_all(&mut *tx)
758 
        .await?;
759 
    sqlx::query(
760 
        "UPDATE flow_jobs SET status = 'cancelled', error = ?, finished_ms = ?, token_hash = ''
761 
         WHERE run_id = ? AND status IN ('queued', 'running')",
762 
    )
763 
    .bind(why)
764 
    .bind(now)
765 
    .bind(run_id)
766 
    .execute(&mut *tx)
767 
    .await?;
768 
    sqlx::query("UPDATE flow_runs SET status = 'cancelled', finished_ms = ? WHERE id = ? AND status IN ('queued', 'running')")
769 
        .bind(now)
770 
        .bind(run_id)
771 
        .execute(&mut *tx)
772 
        .await?;
773 
    tx.commit().await?;
774 
    let mut out = Vec::new();
775 
    for (id,) in ids {
776 
        if let Some(x) = job(pool, id).await? {
777 
            out.push(x);
778 
        }
779 
    }
780 
    Ok(out)
781 
}
782
783 
/// The push runs of `bookmark` still going, older than run `number`.
784 
pub async fn live_push_runs_before(pool: &Pool, repo_id: i64, bookmark: &str, number: i64) -> sqlx::Result<Vec<Run>> {
785 
    sqlx::query_as(
786 
        "SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND number < ? AND trigger = 'push'
787 
         AND status IN ('queued', 'running') ORDER BY id",
788 
    )
789 
    .bind(repo_id)
790 
    .bind(bookmark)
791 
    .bind(number)
792 
    .fetch_all(pool)
793 
    .await
794 
}
795
796 
/// The newest run of exactly this state (heads joined with `+`, sorted).
797 
pub async fn latest_for_heads(pool: &Pool, repo_id: i64, heads: &str) -> sqlx::Result<Option<Run>> {
798 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND heads = ? ORDER BY id DESC LIMIT 1")
799 
        .bind(repo_id)
800 
        .bind(heads)
801 
        .fetch_optional(pool)
802 
        .await
803 
}
804
805 
/// Record what became of job `job_id`'s signed receipt.
806 
pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {
807 
    sqlx::query("UPDATE flow_jobs SET receipt_attest = ?, receipt_signer = ?, receipt_note = ? WHERE id = ?")
808 
        .bind(attest)
809 
        .bind(signer)
810 
        .bind(note)
811 
        .bind(job_id)
812 
        .execute(pool)
813 
        .await?;
814 
    Ok(())
815 
}
816
817 
/// A job's artifact, kept as a blob of the run's repository.
818 
#[derive(Clone, Debug, FromRow)]
819 
pub struct Artifact {
820 
    pub job_id: i64,
821 
    pub name: String,
822 
    pub path: String,
823 
    pub blob: String,
824 
    pub size: i64,
825 
    /// BLAKE3 of the bytes.
826 
    pub content: String,
827 
    pub created_ms: i64,
828 
}
829
830 
pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {
831 
    sqlx::query("INSERT OR REPLACE INTO flow_artifacts (job_id, name, path, blob, size, content, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?)")
832 
        .bind(a.job_id)
833 
        .bind(&a.name)
834 
        .bind(&a.path)
835 
        .bind(&a.blob)
836 
        .bind(a.size)
837 
        .bind(&a.content)
838 
        .bind(a.created_ms)
839 
        .execute(pool)
840 
        .await?;
841 
    Ok(())
842 
}
843
844 
/// Every artifact blob a repository's runs keep, for garbage collection to
845 
/// keep too (#126): the database is the only thing that names them.
846 
pub async fn artifact_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
847 
    sqlx::query_scalar(
848 
        "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 = ?",
849 
    )
850 
    .bind(repo_id)
851 
    .fetch_all(pool)
852 
    .await
853 
}
854
855 
/// Job `job_id`'s artifacts, by name.
856 
pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
857 
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = ? ORDER BY name").bind(job_id).fetch_all(pool).await
858 
}
859
860 
/// Where an artifact with given content came from: the run and job that
861 
/// built it, and the job's signed receipt when one was kept.
862 
#[derive(Clone, Debug, FromRow)]
863 
pub struct Provenance {
864 
    pub run_number: i64,
865 
    pub job: String,
866 
    pub name: String,
867 
    pub receipt_attest: String,
868 
    pub receipt_signer: String,
869 
}
870
871 
/// The build of the bytes whose BLAKE3 is `content` in repository
872 
/// `repo_id`: a passed job, one with a kept signed receipt first, newest
873 
/// run first.
874 
pub async fn provenance(pool: &Pool, repo_id: i64, content: &str) -> sqlx::Result<Option<Provenance>> {
875 
    sqlx::query_as(
876 
        "SELECT r.number AS run_number, j.name AS job, a.name AS name, j.receipt_attest, j.receipt_signer \
877 
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id \
878 
         WHERE r.repo_id = ? AND a.content = ? AND j.status = 'passed' \
879 
         ORDER BY (j.receipt_attest != '') DESC, r.number DESC LIMIT 1",
880 
    )
881 
    .bind(repo_id)
882 
    .bind(content)
883 
    .fetch_optional(pool)
884 
    .await
885 
}