OXYGENHUB
# e1dc108bd814 oxygenhub / crates / hub / src / db / flow.rs
25.4 KiB · Rust · last changed Jobs go to runners of the right architecture 1 month 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, 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 
}
91
92 
impl Job {
93 
    pub fn depends_on(&self) -> Vec<String> {
94 
        serde_json::from_str(&self.depends).unwrap_or_default()
95 
    }
96
97 
    pub fn done(&self) -> bool {
98 
        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error" | "cancelled")
99 
    }
100 
}
101
102 
#[derive(Clone, Debug, FromRow)]
103 
pub struct LogLine {
104 
    pub seq: i64,
105 
    pub t: i64,
106 
    pub stream: String,
107 
    pub text: String,
108 
}
109
110 
/// A job as the trigger creates it.
111 
pub struct NewJob {
112 
    pub name: String,
113 
    pub depth: i64,
114 
    pub depends: Vec<String>,
115 
    pub spec: String,
116 
    pub sandbox: String,
117 
    pub runtime: String,
118 
}
119
120 
pub struct NewRun<'a> {
121 
    pub repo_id: i64,
122 
    pub bookmark: &'a str,
123 
    pub state: &'a str,
124 
    pub heads: &'a str,
125 
    pub pipeline_hash: &'a str,
126 
    pub actor: &'a str,
127 
    /// `push`, `manual` or `rerun:N`.
128 
    pub trigger: &'a str,
129 
    /// Problems when the pipeline is invalid: the run is recorded in error.
130 
    pub error: Option<&'a str>,
131 
}
132
133 
pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {
134 
    let mut tx = super::begin_write(pool).await?;
135 
    let now = now_ms();
136 
    let status = if r.error.is_some() { "error" } else { "queued" };
137 
    let id: i64 = sqlx::query_scalar(
138 
        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \
139 
         VALUES (?, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
140 
    )
141 
    .bind(r.repo_id)
142 
    .bind(r.repo_id)
143 
    .bind(r.bookmark)
144 
    .bind(r.state)
145 
    .bind(r.heads)
146 
    .bind(r.pipeline_hash)
147 
    .bind(status)
148 
    .bind(r.error.unwrap_or(""))
149 
    .bind(r.actor)
150 
    .bind(r.trigger)
151 
    .bind(now)
152 
    .bind(r.error.map(|_| now))
153 
    .fetch_one(&mut *tx)
154 
    .await?;
155 
    for j in jobs {
156 
        sqlx::query(
157 
            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued')",
158 
        )
159 
        .bind(id)
160 
        .bind(&j.name)
161 
        .bind(j.depth)
162 
        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))
163 
        .bind(&j.spec)
164 
        .bind(&j.sandbox)
165 
        .bind(&j.runtime)
166 
        .execute(&mut *tx)
167 
        .await?;
168 
    }
169 
    tx.commit().await?;
170 
    Ok(id)
171 
}
172
173 
/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).
174 
fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {
175 
    let mut out: Vec<String> = [spec, sandbox]
176 
        .iter()
177 
        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())
178 
        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())
179 
        .filter_map(|l| l.as_str().map(String::from))
180 
        .collect();
181 
    out.sort();
182 
    out.dedup();
183 
    out
184 
}
185
186 
impl Job {
187 
    /// The runner labels this job needs.
188 
    pub fn needs(&self) -> Vec<String> {
189 
        labels_in(&self.spec, &self.sandbox)
190 
    }
191 
}
192
193 
/// The architecture a sandbox's JSON asks for, if any.
194 
fn arch_in(sandbox: &str) -> Option<String> {
195 
    serde_json::from_str::<serde_json::Value>(sandbox).ok()?["arch"].as_str().map(String::from)
196 
}
197
198 
/// A job waiting for or held by a runner, with where it comes from.
199 
#[derive(Clone, Debug, FromRow)]
200 
pub struct QueueEntry {
201 
    pub job_id: i64,
202 
    pub run_id: i64,
203 
    pub repo_id: i64,
204 
    pub owner: String,
205 
    pub name: String,
206 
    pub visibility: String,
207 
    pub run_number: i64,
208 
    pub job: String,
209 
    pub status: String,
210 
    /// The runner holding it, when running.
211 
    pub runner: String,
212 
    pub spec: String,
213 
    pub sandbox: String,
214 
}
215
216 
impl QueueEntry {
217 
    /// The runner labels the job needs.
218 
    pub fn needs(&self) -> Vec<String> {
219 
        labels_in(&self.spec, &self.sandbox)
220 
    }
221
222 
    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).
223 
    pub fn arch(&self) -> Option<String> {
224 
        arch_in(&self.sandbox)
225 
    }
226 
}
227
228 
/// Every job of a live run that is running or queued: running ones first,
229 
/// then queued ones in the order runners claim them (hub spec §13.2).
230 
pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
231 
    sqlx::query_as(
232 
        "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 \
233 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
234 
         WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \
235 
         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, j.run_id, j.depth, j.id",
236 
    )
237 
    .fetch_all(pool)
238 
    .await
239 
}
240
241 
/// A repository's runs, newest first, numbered below `before`.
242 
pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
243 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number < ? ORDER BY number DESC LIMIT ?")
244 
        .bind(repo_id)
245 
        .bind(before.unwrap_or(i64::MAX))
246 
        .bind(limit)
247 
        .fetch_all(pool)
248 
        .await
249 
}
250
251 
/// The latest run on this bookmark of this repository, for what a push
252 
/// changed since (a job's scope).
253 
/// The run a push to `bookmark` is scoped against: the newest that was not
254 
/// cancelled, nor a push run still going (a new push supersedes it), so
255 
/// that what a cancelled run would have checked is checked by the next.
256 
pub async fn latest_on(pool: &Pool, repo_id: i64, bookmark: &str) -> sqlx::Result<Option<Run>> {
257 
    sqlx::query_as(
258 
        "SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND heads != '' AND status != 'cancelled'
259 
         AND NOT (trigger = 'push' AND status IN ('queued', 'running')) ORDER BY id DESC LIMIT 1",
260 
    )
261 
    .bind(repo_id)
262 
    .bind(bookmark)
263 
    .fetch_optional(pool)
264 
    .await
265 
}
266
267 
pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
268 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND id < ? ORDER BY id DESC LIMIT ?")
269 
        .bind(repo_id)
270 
        .bind(before.unwrap_or(i64::MAX))
271 
        .bind(limit)
272 
        .fetch_all(pool)
273 
        .await
274 
}
275
276 
/// A repository's run by its number there.
277 
pub async fn run(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<Run>> {
278 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number = ?").bind(repo_id).bind(number).fetch_optional(pool).await
279 
}
280
281 
pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {
282 
    sqlx::query_as("SELECT * FROM flow_runs WHERE id = ?").bind(id).fetch_optional(pool).await
283 
}
284
285 
pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {
286 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ? ORDER BY depth, name").bind(run_id).fetch_all(pool).await
287 
}
288
289 
pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {
290 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(id).fetch_optional(pool).await
291 
}
292
293 
/// The running job a job token belongs to, while it may still be used.
294 
pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {
295 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = ? AND status = 'running' AND started_ms >= ?")
296 
        .bind(token_hash)
297 
        .bind(now_ms() - max_age_ms)
298 
        .fetch_optional(pool)
299 
        .await
300 
}
301
302 
/// Take the oldest queued job whose dependencies all passed and whose tier
303 
/// the runner offers, and mark it running for `runner` with a token.
304 
/// Let a job's token also read these repositories (its pinned subrepos).
305 
pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {
306 
    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();
307 
    sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;
308 
    Ok(())
309 
}
310
311 
impl Job {
312 
    pub fn also_read_ids(&self) -> Vec<i64> {
313 
        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()
314 
    }
315 
}
316
317 
/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no
318 
/// limit, and the rootfs images it holds, `None` when it does not say.
319 
#[derive(Clone, Debug, Default)]
320 
pub struct Room {
321 
    pub memory_mib: Option<u32>,
322 
    pub cpus: Option<u32>,
323 
    pub images: Option<std::collections::HashSet<String>>,
324 
    /// The runner's architecture (`x86_64`, `riscv64`, …), empty when unknown.
325 
    pub arch: String,
326 
    /// Each image's architecture as runners have reported it (`flow_images`).
327 
    pub image_archs: std::collections::HashMap<String, String>,
328 
}
329
330 
impl Room {
331 
    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,
332 
    /// and its rootfs image among the runner's (oxygenhub#73).
333 
    pub fn fits(&self, sandbox_json: &str) -> bool {
334 
        let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();
335 
        let need = |k: &str| v[k].as_u64().unwrap_or(0);
336 
        let rootfs = v["rootfs"].as_str().unwrap_or("");
337 
        // The architecture: the sandbox's own ask, else the image's as known;
338 
        // an unknown one matches, as before (OxygenFlow spec §11.3).
339 
        let wanted = v["arch"].as_str().map(String::from).or_else(|| self.image_archs.get(rootfs).cloned());
340 
        let arch_ok = self.arch.is_empty() || wanted.as_deref().is_none_or(|w| w == self.arch);
341 
        arch_ok
342 
            && self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)
343 
            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)
344 
            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)))
345 
    }
346 
}
347
348 
/// Remember which architecture each image is for, as a runner reports it.
349 
pub async fn record_image_archs(pool: &Pool, archs: &std::collections::BTreeMap<String, String>) -> sqlx::Result<()> {
350 
    for (hash, arch) in archs.iter().take(500) {
351 
        if hash.len() != 64 || !flow_core::arch_ok(arch) {
352 
            continue;
353 
        }
354 
        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")
355 
        .bind(hash.to_ascii_lowercase())
356 
        .bind(arch)
357 
        .bind(now_ms())
358 
        .execute(pool)
359 
        .await?;
360 
    }
361 
    Ok(())
362 
}
363
364 
/// Every image's architecture the hub knows.
365 
pub async fn image_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
366 
    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, arch FROM flow_images").fetch_all(pool).await?;
367 
    Ok(rows.into_iter().collect())
368 
}
369
370 
pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {
371 
    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await
372 
}
373
374 
/// The oldest runnable job that fits in `room`.
375 
/// The first queued job a runner with these tiers, this room and this
376 
/// scope may take, now running under `runner`. A scoped runner never sees
377 
/// another scope's jobs (oxygenhub#41).
378 
pub async fn claim_within(
379 
    pool: &Pool,
380 
    runtimes: &[String],
381 
    runner: &str,
382 
    token_hash: &str,
383 
    room: Room,
384 
    scope: &super::runners::Scope,
385 
    labels: &[String],
386 
) -> sqlx::Result<Option<Job>> {
387 
    let mut tx = super::begin_write(pool).await?;
388 
    let queued: Vec<Job> = sqlx::query_as(
389 
        "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 \
390 
         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') \
391 
         AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \
392 
         ORDER BY j.run_id, j.depth, j.id LIMIT 200",
393 
    )
394 
    .bind(&scope.owner)
395 
    .bind(&scope.owner)
396 
    .bind(scope.repo_id)
397 
    .bind(scope.repo_id)
398 
    .fetch_all(&mut *tx)
399 
    .await?;
400 
    for j in queued {
401 
        if !runtimes.iter().any(|r| r == &j.runtime) || !room.fits(&j.sandbox) {
402 
            continue;
403 
        }
404 
        // Every label the job and its sandbox ask for (`runs-on`).
405 
        if !j.needs().iter().all(|l| labels.contains(l)) {
406 
            continue;
407 
        }
408 
        let deps = j.depends_on();
409 
        if !deps.is_empty() {
410 
            let mut passed = 0i64;
411 
            for d in &deps {
412 
                let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")
413 
                    .bind(j.run_id)
414 
                    .bind(d)
415 
                    .fetch_optional(&mut *tx)
416 
                    .await?;
417 
                if st.as_deref() == Some("passed") {
418 
                    passed += 1;
419 
                }
420 
            }
421 
            if passed != deps.len() as i64 {
422 
                continue;
423 
            }
424 
        }
425 
        let now = now_ms();
426 
        let done = sqlx::query(
427 
            "UPDATE flow_jobs SET status = 'running', runner = ?, token_hash = ?, started_ms = ? WHERE id = ? AND status = 'queued'",
428 
        )
429 
        .bind(runner)
430 
        .bind(token_hash)
431 
        .bind(now)
432 
        .bind(j.id)
433 
        .execute(&mut *tx)
434 
        .await?;
435 
        if done.rows_affected() == 1 {
436 
            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = ? AND status = 'queued'")
437 
                .bind(j.run_id)
438 
                .execute(&mut *tx)
439 
                .await?;
440 
            tx.commit().await?;
441 
            return job(pool, j.id).await;
442 
        }
443 
    }
444 
    tx.commit().await?;
445 
    Ok(None)
446 
}
447
448 
/// Append records; returns how many were kept (the rest exceed the cap).
449 
pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {
450 
    let mut tx = super::begin_write(pool).await?;
451 
    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
452 
    let room = (cap - count).max(0) as usize;
453 
    let keep = &records[..records.len().min(room)];
454 
    for (i, r) in keep.iter().enumerate() {
455 
        sqlx::query("INSERT INTO flow_logs (job_id, seq, t, stream, text) VALUES (?, ?, ?, ?, ?)")
456 
            .bind(job_id)
457 
            .bind(count + i as i64 + 1)
458 
            .bind(r.t as i64)
459 
            .bind(&r.stream)
460 
            .bind(&r.text)
461 
            .execute(&mut *tx)
462 
            .await?;
463 
    }
464 
    sqlx::query("UPDATE flow_jobs SET log_count = log_count + ? WHERE id = ?")
465 
        .bind(keep.len() as i64)
466 
        .bind(job_id)
467 
        .execute(&mut *tx)
468 
        .await?;
469 
    tx.commit().await?;
470 
    Ok(keep.len())
471 
}
472
473 
pub async fn log(pool: &Pool, job_id: i64, after: i64, limit: i64) -> sqlx::Result<Vec<LogLine>> {
474 
    sqlx::query_as("SELECT seq, t, stream, text FROM flow_logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT ?")
475 
        .bind(job_id)
476 
        .bind(after)
477 
        .bind(limit)
478 
        .fetch_all(pool)
479 
        .await
480 
}
481
482 
/// Jobs running since before `ms` (the oldest first).
483 
/// Jobs still running at `now` past their own time limit (`default_s` for
484 
/// one claimed without) plus `grace_ms`.
485 
pub async fn running_past_limit(pool: &Pool, now: i64, default_s: u64, grace_ms: i64) -> sqlx::Result<Vec<Job>> {
486 
    sqlx::query_as(
487 
        "SELECT * FROM flow_jobs WHERE status = 'running' \
488 
         AND COALESCE(started_ms, 0) + (CASE WHEN timeout_s > 0 THEN timeout_s ELSE ? END) * 1000 + ? < ? ORDER BY started_ms",
489 
    )
490 
    .bind(default_s as i64)
491 
    .bind(grace_ms)
492 
    .bind(now)
493 
    .fetch_all(pool)
494 
    .await
495 
}
496
497 
/// Record the limit a claimed job runs under.
498 
pub async fn set_job_timeout(pool: &Pool, job_id: i64, timeout_s: u64) -> sqlx::Result<()> {
499 
    sqlx::query("UPDATE flow_jobs SET timeout_s = ? WHERE id = ?").bind(timeout_s as i64).bind(job_id).execute(pool).await?;
500 
    Ok(())
501 
}
502
503 
pub async fn running_since_before(pool: &Pool, ms: i64) -> sqlx::Result<Vec<Job>> {
504 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < ? ORDER BY started_ms")
505 
        .bind(ms)
506 
        .fetch_all(pool)
507 
        .await
508 
}
509
510 
/// The jobs running on the runner with this id.
511 
pub async fn running_on(pool: &Pool, runner_id: i64) -> sqlx::Result<Vec<Job>> {
512 
    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = ? ORDER BY id").bind(runner_id).fetch_all(pool).await
513 
}
514
515 
pub struct Outcome<'a> {
516 
    pub status: &'a str,
517 
    pub exit_code: Option<i64>,
518 
    pub error: &'a str,
519 
    pub receipt: &'a str,
520 
    pub receipt_hash: &'a str,
521 
}
522
523 
/// Record a job's end, skip the jobs that can no longer run, and settle the
524 
/// run. Returns the jobs whose status changed (for their checks).
525 
pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {
526 
    let mut tx = super::begin_write(pool).await?;
527 
    let now = now_ms();
528 
    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;
529 
    sqlx::query("UPDATE flow_jobs SET status = ?, exit_code = ?, error = ?, receipt = ?, receipt_hash = ?, finished_ms = ?, token_hash = '' WHERE id = ?")
530 
        .bind(o.status)
531 
        .bind(o.exit_code)
532 
        .bind(o.error)
533 
        .bind(o.receipt)
534 
        .bind(o.receipt_hash)
535 
        .bind(now)
536 
        .bind(job_id)
537 
        .execute(&mut *tx)
538 
        .await?;
539 
    let mut changed = vec![job_id];
540 
    // Dependents of anything that did not pass are skipped, transitively.
541 
    loop {
542 
        let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
543 
        let blocked: Vec<i64> = all
544 
            .iter()
545 
            .filter(|x| x.status == "queued")
546 
            .filter(|x| {
547 
                x.depends_on()
548 
                    .iter()
549 
                    .any(|d| all.iter().any(|y| &y.name == d && matches!(y.status.as_str(), "failed" | "skipped" | "error")))
550 
            })
551 
            .map(|x| x.id)
552 
            .collect();
553 
        if blocked.is_empty() {
554 
            break;
555 
        }
556 
        for id in blocked {
557 
            sqlx::query("UPDATE flow_jobs SET status = 'skipped', finished_ms = ? WHERE id = ?")
558 
                .bind(now)
559 
                .bind(id)
560 
                .execute(&mut *tx)
561 
                .await?;
562 
            changed.push(id);
563 
        }
564 
    }
565 
    let all: Vec<Job> = sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ?").bind(j.run_id).fetch_all(&mut *tx).await?;
566 
    if all.iter().all(Job::done) {
567 
        let status = if all.iter().all(|x| x.status == "passed") { "passed" } else { "failed" };
568 
        sqlx::query("UPDATE flow_runs SET status = ?, finished_ms = ? WHERE id = ?")
569 
            .bind(status)
570 
            .bind(now)
571 
            .bind(j.run_id)
572 
            .execute(&mut *tx)
573 
            .await?;
574 
    }
575 
    tx.commit().await?;
576 
    let mut out = Vec::new();
577 
    for id in changed {
578 
        if let Some(x) = job(pool, id).await? {
579 
            out.push(x);
580 
        }
581 
    }
582 
    Ok(out)
583 
}
584
585 
/// Cancel a run (oxygenflow#32): its jobs not yet done end as cancelled,
586 
/// and so does the run. A runner learns of it when the hub next refuses
587 
/// its log lines, and stops the job. Returns the jobs whose status changed.
588 
pub async fn cancel_run(pool: &Pool, run_id: i64, why: &str) -> sqlx::Result<Vec<Job>> {
589 
    let now = now_ms();
590 
    let mut tx = pool.begin().await?;
591 
    let ids: Vec<(i64,)> = sqlx::query_as("SELECT id FROM flow_jobs WHERE run_id = ? AND status IN ('queued', 'running')")
592 
        .bind(run_id)
593 
        .fetch_all(&mut *tx)
594 
        .await?;
595 
    sqlx::query(
596 
        "UPDATE flow_jobs SET status = 'cancelled', error = ?, finished_ms = ?, token_hash = ''
597 
         WHERE run_id = ? AND status IN ('queued', 'running')",
598 
    )
599 
    .bind(why)
600 
    .bind(now)
601 
    .bind(run_id)
602 
    .execute(&mut *tx)
603 
    .await?;
604 
    sqlx::query("UPDATE flow_runs SET status = 'cancelled', finished_ms = ? WHERE id = ? AND status IN ('queued', 'running')")
605 
        .bind(now)
606 
        .bind(run_id)
607 
        .execute(&mut *tx)
608 
        .await?;
609 
    tx.commit().await?;
610 
    let mut out = Vec::new();
611 
    for (id,) in ids {
612 
        if let Some(x) = job(pool, id).await? {
613 
            out.push(x);
614 
        }
615 
    }
616 
    Ok(out)
617 
}
618
619 
/// The push runs of `bookmark` still going, older than run `number`.
620 
pub async fn live_push_runs_before(pool: &Pool, repo_id: i64, bookmark: &str, number: i64) -> sqlx::Result<Vec<Run>> {
621 
    sqlx::query_as(
622 
        "SELECT * FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND number < ? AND trigger = 'push'
623 
         AND status IN ('queued', 'running') ORDER BY id",
624 
    )
625 
    .bind(repo_id)
626 
    .bind(bookmark)
627 
    .bind(number)
628 
    .fetch_all(pool)
629 
    .await
630 
}
631
632 
/// The newest run of exactly this state (heads joined with `+`, sorted).
633 
pub async fn latest_for_heads(pool: &Pool, repo_id: i64, heads: &str) -> sqlx::Result<Option<Run>> {
634 
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND heads = ? ORDER BY id DESC LIMIT 1")
635 
        .bind(repo_id)
636 
        .bind(heads)
637 
        .fetch_optional(pool)
638 
        .await
639 
}
640
641 
/// Record what became of job `job_id`'s signed receipt.
642 
pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {
643 
    sqlx::query("UPDATE flow_jobs SET receipt_attest = ?, receipt_signer = ?, receipt_note = ? WHERE id = ?")
644 
        .bind(attest)
645 
        .bind(signer)
646 
        .bind(note)
647 
        .bind(job_id)
648 
        .execute(pool)
649 
        .await?;
650 
    Ok(())
651 
}
652
653 
/// A job's artifact, kept as a blob of the run's repository.
654 
#[derive(Clone, Debug, FromRow)]
655 
pub struct Artifact {
656 
    pub job_id: i64,
657 
    pub name: String,
658 
    pub path: String,
659 
    pub blob: String,
660 
    pub size: i64,
661 
    /// BLAKE3 of the bytes.
662 
    pub content: String,
663 
    pub created_ms: i64,
664 
}
665
666 
pub async fn add_artifact(pool: &Pool, a: &Artifact) -> sqlx::Result<()> {
667 
    sqlx::query("INSERT OR REPLACE INTO flow_artifacts (job_id, name, path, blob, size, content, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?)")
668 
        .bind(a.job_id)
669 
        .bind(&a.name)
670 
        .bind(&a.path)
671 
        .bind(&a.blob)
672 
        .bind(a.size)
673 
        .bind(&a.content)
674 
        .bind(a.created_ms)
675 
        .execute(pool)
676 
        .await?;
677 
    Ok(())
678 
}
679
680 
/// Job `job_id`'s artifacts, by name.
681 
pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
682 
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = ? ORDER BY name").bind(job_id).fetch_all(pool).await
683 
}
684
685 
/// Where an artifact with given content came from: the run and job that
686 
/// built it, and the job's signed receipt when one was kept.
687 
#[derive(Clone, Debug, FromRow)]
688 
pub struct Provenance {
689 
    pub run_number: i64,
690 
    pub job: String,
691 
    pub name: String,
692 
    pub receipt_attest: String,
693 
    pub receipt_signer: String,
694 
}
695
696 
/// The build of the bytes whose BLAKE3 is `content` in repository
697 
/// `repo_id`: a passed job, one with a kept signed receipt first, newest
698 
/// run first.
699 
pub async fn provenance(pool: &Pool, repo_id: i64, content: &str) -> sqlx::Result<Option<Provenance>> {
700 
    sqlx::query_as(
701 
        "SELECT r.number AS run_number, j.name AS job, a.name AS name, j.receipt_attest, j.receipt_signer \
702 
         FROM flow_artifacts a JOIN flow_jobs j ON j.id = a.job_id JOIN flow_runs r ON r.id = j.run_id \
703 
         WHERE r.repo_id = ? AND a.content = ? AND j.status = 'passed' \
704 
         ORDER BY (j.receipt_attest != '') DESC, r.number DESC LIMIT 1",
705 
    )
706 
    .bind(repo_id)
707 
    .bind(content)
708 
    .fetch_optional(pool)
709 
    .await
710 
}