OXYGENHUB

Reports: keep a job's report file by file, index its declaration, metrics and files a36bab6b9ffc

A Flow job's report output arrives as its files, each kept as a blob by
POST /api/v1/flow/jobs/{job}/reports/{name}/files, then its manifest as
the report's artifact. The hub checks the manifest and that it holds
every file, reads report.toml (kind, title, summary, metrics with units
and thresholds, a free-form context table) and indexes reports,
report_metrics and report_files (migration 0050). Garbage collection
keeps every file a report names. Bumps deps/oxygenflow for the report
output kind.

refs #127
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-10-10 23:56 UTC
Signature
✔ signed by Andy McCall
Hash
a36bab6b9ffcbbac5e9f2500e4938f546d17993ef51f799e351695986871128f
Depends on
0b203bbf7cc3 380996db84e5 42fa95540468 4597750166e7 50d93fa443dc 554ccacfc123 8a3334b7a364 a3c8288b0141 d23464c0ac60 d29070faef66 ef20fdc14e8b
Depended on by
40a43921ef42 fac6853c4e46 e4203e141e90 3b1bc27fe805 5d90e0e045e2 cd437c4f5f8d 20b8fe85888b 92a797b20912 c8df33a6829f
In bookmarks
api-unknown-fields cancel-landed-runs clear-waiting close-proposal-140 diffusion-pin-bookmark draft-ready image-check job-waits main oh-comment-body-file proposal-list-fast run-one-job wiki-says-so
In tags
v0.4.2 v0.5.0 v0.5.1 v0.5.2 v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
Edits
17 atoms, 2 tree operations, 10 files

Browse the tree at this change

added crates/hub/migrations/0050_reports.sql +44 −0
1 +
-- Reports (oxygenhub#127, docs/reports.md): a directory a Flow job writes
2 +
-- with report.toml at its top. Its files are blobs of the repository and
3 +
-- its manifest is the job's artifact of that name; these tables index it
4 +
-- for the Reports tab, its history and checks, and name every file blob
5 +
-- so garbage collection keeps it.
6 +
CREATE TABLE reports (
7 +
    id          INTEGER PRIMARY KEY,
8 +
    repo_id     INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,
9 +
    name        TEXT NOT NULL,
10 +
    kind        TEXT NOT NULL,
11 +
    title       TEXT NOT NULL,
12 +
    summary     TEXT NOT NULL,
13 +
    context     TEXT NOT NULL,
14 +
    run_id      INTEGER REFERENCES flow_runs(id) ON DELETE CASCADE,
15 +
    job_id      INTEGER,
16 +
    run_number  INTEGER,
17 +
    job         TEXT NOT NULL,
18 +
    bookmark    TEXT NOT NULL,
19 +
    state       TEXT NOT NULL,
20 +
    manifest    TEXT,
21 +
    files       INTEGER NOT NULL,
22 +
    bytes       INTEGER NOT NULL,
23 +
    created_ms  INTEGER NOT NULL
24 +
);
25 +
CREATE UNIQUE INDEX reports_job ON reports (job_id, name);
26 +
CREATE INDEX reports_history ON reports (repo_id, name, bookmark, created_ms);
27 +
CREATE TABLE report_metrics (
28 +
    report_id   INTEGER NOT NULL REFERENCES reports(id) ON DELETE CASCADE,
29 +
    key         TEXT NOT NULL,
30 +
    value       REAL NOT NULL,
31 +
    unit        TEXT NOT NULL,
32 +
    better      TEXT NOT NULL,
33 +
    fail_above  REAL,
34 +
    fail_below  REAL,
35 +
    PRIMARY KEY (report_id, key)
36 +
);
37 +
CREATE TABLE report_files (
38 +
    report_id   INTEGER NOT NULL REFERENCES reports(id) ON DELETE CASCADE,
39 +
    path        TEXT NOT NULL,
40 +
    blob        TEXT NOT NULL,
41 +
    size        INTEGER NOT NULL,
42 +
    PRIMARY KEY (report_id, path)
43 +
);
44 +
CREATE INDEX report_files_blob ON report_files (blob);
added crates/hub/src/reports.rs +382 −0
1 +
//! Reports (oxygenhub#127, docs/reports.md): a directory a Flow job writes
2 +
//! with `report.toml` at its top. The runner uploads each file as a blob of
3 +
//! the run's repository, then a manifest naming them as the job's artifact
4 +
//! under the report's name. The hub checks the manifest, reads the
5 +
//! declaration and indexes the report, its metrics and its files, so the
6 +
//! Reports tab, its history and checks read only the database, and garbage
7 +
//! collection keeps every file a kept report names.
8 +
9 +
use crate::db::{now_ms, Pool};
10 +
use crate::Shared;
11 +
use serde::Deserialize;
12 +
13 +
/// Files in one report, and their total size; the runner keeps to these.
14 +
pub const MAX_FILES: usize = 2000;
15 +
pub const MAX_BYTES: u64 = 512 << 20;
16 +
/// Largest `report.toml` read.
17 +
const MAX_DECLARATION: usize = 64 << 10;
18 +
/// Metrics kept from one report.
19 +
const MAX_METRICS: usize = 200;
20 +
/// The kinds a report may declare.
21 +
pub const KINDS: [&str; 4] = ["coverage", "tests", "performance", "custom"];
22 +
23 +
/// A report file's path: `/`-separated names of letters, digits, `.`,
24 +
/// `_`, `-` and `+`, never `.` or `..`, so it travels in a header and a URL
25 +
/// as it is.
26 +
pub fn path_ok(p: &str) -> bool {
27 +
    p.len() <= 1024
28 +
        && p.split('/').count() <= 32
29 +
        && p.split('/').all(|n| {
30 +
            !n.is_empty() && n != "." && n != ".." && n.len() <= 255 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-+".contains(&b))
31 +
        })
32 +
}
33 +
34 +
/// A report's manifest, as the runner uploads it.
35 +
#[derive(Clone, Debug, Deserialize)]
36 +
pub struct Manifest {
37 +
    pub version: u32,
38 +
    pub files: Vec<ManifestFile>,
39 +
}
40 +
41 +
#[derive(Clone, Debug, Deserialize)]
42 +
pub struct ManifestFile {
43 +
    pub path: String,
44 +
    pub blob: String,
45 +
    pub size: u64,
46 +
}
47 +
48 +
impl Manifest {
49 +
    pub fn bytes(&self) -> u64 {
50 +
        self.files.iter().map(|f| f.size).sum()
51 +
    }
52 +
53 +
    /// The blob of the file at `path`.
54 +
    pub fn blob(&self, path: &str) -> Option<&str> {
55 +
        self.files.iter().find(|f| f.path == path).map(|f| f.blob.as_str())
56 +
    }
57 +
}
58 +
59 +
/// Read and check a manifest: version 1, clean unique paths, blob names,
60 +
/// within the limits, with `report.toml`.
61 +
pub fn parse_manifest(bytes: &[u8]) -> Result<Manifest, String> {
62 +
    let m: Manifest = serde_json::from_slice(bytes).map_err(|e| format!("the report manifest is not valid: {e}"))?;
63 +
    if m.version != 1 {
64 +
        return Err(format!("report manifest version {} is not known; this hub reads version 1", m.version));
65 +
    }
66 +
    if m.files.len() > MAX_FILES {
67 +
        return Err(format!("the report holds {} files; at most {MAX_FILES}", m.files.len()));
68 +
    }
69 +
    if m.bytes() > MAX_BYTES {
70 +
        return Err(format!("the report holds {} MiB; at most {} MiB", m.bytes() >> 20, MAX_BYTES >> 20));
71 +
    }
72 +
    let mut seen = std::collections::HashSet::new();
73 +
    for f in &m.files {
74 +
        if !path_ok(&f.path) {
75 +
            return Err(format!("report file {:?}: names may hold letters, digits, '.', '_', '-' and '+'", f.path));
76 +
        }
77 +
        if !seen.insert(f.path.as_str()) {
78 +
            return Err(format!("report file {} is named twice", f.path));
79 +
        }
80 +
        if oxy_core::hash::ObjectRef::from_hex(&f.blob).is_none() {
81 +
            return Err(format!("report file {}: {:?} is not a blob name", f.path, f.blob));
82 +
        }
83 +
    }
84 +
    if m.blob("report.toml").is_none() {
85 +
        return Err("the report has no report.toml at its top".into());
86 +
    }
87 +
    Ok(m)
88 +
}
89 +
90 +
/// What `report.toml` declares.
91 +
#[derive(Clone, Debug, PartialEq)]
92 +
pub struct Declaration {
93 +
    pub kind: String,
94 +
    pub title: String,
95 +
    pub summary: String,
96 +
    /// The data file of a built-in kind (`lcov.info`, `junit.xml`).
97 +
    pub file: Option<String>,
98 +
    pub metrics: Vec<Metric>,
99 +
    /// The free-form `[context]` table, as JSON.
100 +
    pub context: serde_json::Value,
101 +
}
102 +
103 +
/// A number the hub tracks across runs.
104 +
#[derive(Clone, Debug, PartialEq, sqlx::FromRow)]
105 +
pub struct Metric {
106 +
    pub key: String,
107 +
    pub value: f64,
108 +
    pub unit: String,
109 +
    /// `higher`, `lower`, or empty when neither is better.
110 +
    pub better: String,
111 +
    pub fail_above: Option<f64>,
112 +
    pub fail_below: Option<f64>,
113 +
}
114 +
115 +
impl Declaration {
116 +
    /// The declaration of a report whose `report.toml` could not be read:
117 +
    /// kept as a custom report saying why, so its files stay and the reader
118 +
    /// sees what to fix.
119 +
    pub fn unreadable(name: &str, why: &str) -> Declaration {
120 +
        Declaration {
121 +
            kind: "custom".into(),
122 +
            title: name.into(),
123 +
            summary: format!("report.toml: {why}"),
124 +
            file: None,
125 +
            metrics: vec![],
126 +
            context: serde_json::json!({}),
127 +
        }
128 +
    }
129 +
}
130 +
131 +
fn number(v: &toml::Value) -> Option<f64> {
132 +
    match v {
133 +
        toml::Value::Integer(i) => Some(*i as f64),
134 +
        toml::Value::Float(f) if f.is_finite() => Some(*f),
135 +
        _ => None,
136 +
    }
137 +
}
138 +
139 +
fn metric_key_ok(k: &str) -> bool {
140 +
    !k.is_empty() && k.len() <= 64 && k.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b))
141 +
}
142 +
143 +
/// Read `report.toml` for report `name`; the error says what to fix.
144 +
pub fn parse_declaration(name: &str, bytes: &[u8]) -> Result<Declaration, String> {
145 +
    if bytes.len() > MAX_DECLARATION {
146 +
        return Err(format!("{} KiB; at most {} KiB", bytes.len() >> 10, MAX_DECLARATION >> 10));
147 +
    }
148 +
    let text = std::str::from_utf8(bytes).map_err(|_| "not UTF-8 text".to_string())?;
149 +
    let table: toml::Table = toml::from_str(text).map_err(|e| {
150 +
        let line = e.span().map(|s| text[..s.start.min(text.len())].lines().count().max(1));
151 +
        let what = if e.message().trim().is_empty() { "not valid TOML" } else { e.message().trim() };
152 +
        match line {
153 +
            Some(n) => format!("line {n}: {what}"),
154 +
            None => what.to_string(),
155 +
        }
156 +
    })?;
157 +
    let text_of = |key: &str, max: usize| -> Result<Option<String>, String> {
158 +
        match table.get(key) {
159 +
            None => Ok(None),
160 +
            Some(toml::Value::String(s)) if s.len() <= max => Ok(Some(s.clone())),
161 +
            Some(toml::Value::String(_)) => Err(format!("`{key}` is longer than {max} bytes")),
162 +
            Some(_) => Err(format!("`{key}` is a string")),
163 +
        }
164 +
    };
165 +
    for key in table.keys() {
166 +
        if !matches!(key.as_str(), "kind" | "title" | "summary" | "file" | "metrics" | "context") {
167 +
            return Err(format!("`{key}` is not known (kind, title, summary, file, metrics, context)"));
168 +
        }
169 +
    }
170 +
    let kind = text_of("kind", 32)?.ok_or("`kind` is required: one of coverage, tests, performance, custom")?;
171 +
    if !KINDS.contains(&kind.as_str()) {
172 +
        return Err(format!("kind {kind:?} is not known: one of coverage, tests, performance, custom"));
173 +
    }
174 +
    let file = text_of("file", 1024)?;
175 +
    if let Some(f) = &file {
176 +
        if !path_ok(f) {
177 +
            return Err(format!("file {f:?} is not a path in the report"));
178 +
        }
179 +
    }
180 +
    let mut metrics = Vec::new();
181 +
    match table.get("metrics") {
182 +
        None => {}
183 +
        Some(toml::Value::Table(t)) => {
184 +
            if t.len() > MAX_METRICS {
185 +
                return Err(format!("{} metrics; at most {MAX_METRICS}", t.len()));
186 +
            }
187 +
            for (key, v) in t {
188 +
                if !metric_key_ok(key) {
189 +
                    return Err(format!("metric {key:?}: names hold letters, digits, '.', '_' and '-'"));
190 +
                }
191 +
                metrics.push(parse_metric(key, v).map_err(|e| format!("metric {key}: {e}"))?);
192 +
            }
193 +
        }
194 +
        Some(_) => return Err("`metrics` is a table".into()),
195 +
    }
196 +
    let context = match table.get("context") {
197 +
        None => serde_json::json!({}),
198 +
        Some(v @ toml::Value::Table(_)) => serde_json::to_value(v).map_err(|e| format!("context: {e}"))?,
199 +
        Some(_) => return Err("`context` is a table".into()),
200 +
    };
201 +
    Ok(Declaration {
202 +
        kind,
203 +
        title: text_of("title", 200)?.unwrap_or_else(|| name.to_string()),
204 +
        summary: text_of("summary", 1000)?.unwrap_or_default(),
205 +
        file,
206 +
        metrics,
207 +
        context,
208 +
    })
209 +
}
210 +
211 +
fn parse_metric(key: &str, v: &toml::Value) -> Result<Metric, String> {
212 +
    let mut m = Metric { key: key.into(), value: 0.0, unit: String::new(), better: String::new(), fail_above: None, fail_below: None };
213 +
    if let Some(n) = number(v) {
214 +
        m.value = n;
215 +
        return Ok(m);
216 +
    }
217 +
    let toml::Value::Table(t) = v else {
218 +
        return Err("a number, or a table with `value`".into());
219 +
    };
220 +
    for (k, v) in t {
221 +
        match k.as_str() {
222 +
            "value" => m.value = number(v).ok_or("`value` is a number")?,
223 +
            "unit" => match v {
224 +
                toml::Value::String(s) if s.len() <= 32 => m.unit = s.clone(),
225 +
                _ => return Err("`unit` is a short string".into()),
226 +
            },
227 +
            "better" => match v.as_str() {
228 +
                Some(s @ ("higher" | "lower")) => m.better = s.into(),
229 +
                _ => return Err("`better` is \"higher\" or \"lower\"".into()),
230 +
            },
231 +
            "fail_above" => m.fail_above = Some(number(v).ok_or("`fail_above` is a number")?),
232 +
            "fail_below" => m.fail_below = Some(number(v).ok_or("`fail_below` is a number")?),
233 +
            _ => return Err(format!("`{k}` is not known (value, unit, better, fail_above, fail_below)")),
234 +
        }
235 +
    }
236 +
    if !t.contains_key("value") {
237 +
        return Err("`value` is required".into());
238 +
    }
239 +
    Ok(m)
240 +
}
241 +
242 +
/// Where a report came from.
243 +
pub struct Origin<'a> {
244 +
    pub repo_id: i64,
245 +
    pub run_id: i64,
246 +
    pub run_number: i64,
247 +
    pub job_id: i64,
248 +
    pub job: &'a str,
249 +
    pub bookmark: &'a str,
250 +
    pub state: &'a str,
251 +
}
252 +
253 +
/// Index report `name` of a job, replacing any the job kept before.
254 +
pub async fn keep(app: &Shared, o: &Origin<'_>, name: &str, d: &Declaration, m: &Manifest, manifest: &str) -> sqlx::Result<i64> {
255 +
    let mut tx = crate::db::begin_write(&app.db).await?;
256 +
    sqlx::query("DELETE FROM reports WHERE job_id = ? AND name = ?").bind(o.job_id).bind(name).execute(&mut *tx).await?;
257 +
    let id: i64 = sqlx::query_scalar(
258 +
        "INSERT INTO reports (repo_id, name, kind, title, summary, context, run_id, job_id, run_number, job, bookmark, state, \
259 +
         manifest, files, bytes, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
260 +
    )
261 +
    .bind(o.repo_id)
262 +
    .bind(name)
263 +
    .bind(&d.kind)
264 +
    .bind(&d.title)
265 +
    .bind(&d.summary)
266 +
    .bind(d.context.to_string())
267 +
    .bind(o.run_id)
268 +
    .bind(o.job_id)
269 +
    .bind(o.run_number)
270 +
    .bind(o.job)
271 +
    .bind(o.bookmark)
272 +
    .bind(o.state)
273 +
    .bind(manifest)
274 +
    .bind(m.files.len() as i64)
275 +
    .bind(m.bytes() as i64)
276 +
    .bind(now_ms())
277 +
    .fetch_one(&mut *tx)
278 +
    .await?;
279 +
    for x in &d.metrics {
280 +
        sqlx::query(
281 +
            "INSERT INTO report_metrics (report_id, key, value, unit, better, fail_above, fail_below) VALUES (?, ?, ?, ?, ?, ?, ?)",
282 +
        )
283 +
        .bind(id)
284 +
        .bind(&x.key)
285 +
        .bind(x.value)
286 +
        .bind(&x.unit)
287 +
        .bind(&x.better)
288 +
        .bind(x.fail_above)
289 +
        .bind(x.fail_below)
290 +
        .execute(&mut *tx)
291 +
        .await?;
292 +
    }
293 +
    for f in &m.files {
294 +
        sqlx::query("INSERT INTO report_files (report_id, path, blob, size) VALUES (?, ?, ?, ?)")
295 +
            .bind(id)
296 +
            .bind(&f.path)
297 +
            .bind(&f.blob)
298 +
            .bind(f.size as i64)
299 +
            .execute(&mut *tx)
300 +
            .await?;
301 +
    }
302 +
    tx.commit().await?;
303 +
    Ok(id)
304 +
}
305 +
306 +
/// Report `id`'s metrics, by key.
307 +
pub async fn metrics(pool: &Pool, id: i64) -> sqlx::Result<Vec<Metric>> {
308 +
    sqlx::query_as("SELECT key, value, unit, better, fail_above, fail_below FROM report_metrics WHERE report_id = ? ORDER BY key")
309 +
        .bind(id)
310 +
        .fetch_all(pool)
311 +
        .await
312 +
}
313 +
314 +
/// Every file blob a repository's kept reports name, for garbage collection
315 +
/// to keep too: the database is the only thing that names them.
316 +
pub async fn file_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
317 +
    sqlx::query_scalar("SELECT DISTINCT f.blob FROM report_files f JOIN reports r ON r.id = f.report_id WHERE r.repo_id = ?")
318 +
        .bind(repo_id)
319 +
        .fetch_all(pool)
320 +
        .await
321 +
}
322 +
323 +
#[cfg(test)]
324 +
mod tests {
325 +
    use super::*;
326 +
327 +
    #[test]
328 +
    fn declarations_are_read_with_their_metrics_and_context() {
329 +
        let d = parse_declaration(
330 +
            "frame-timing",
331 +
            br#"
332 +
kind    = "performance"
333 +
title   = "Frame timing"
334 +
summary = "Median frame 16.4 ms"
335 +
336 +
[metrics]
337 +
median_frame_ms = { value = 16.4, unit = "ms", better = "lower" }
338 +
p99_frame_ms    = { value = 18.1, unit = "ms", better = "lower", fail_above = 20 }
339 +
scenes          = 12
340 +
341 +
[context]
342 +
target_fps    = 30
343 +
budget_cycles = 268800
344 +
"#,
345 +
        )
346 +
        .unwrap();
347 +
        assert_eq!((d.kind.as_str(), d.title.as_str()), ("performance", "Frame timing"));
348 +
        let keys: Vec<&str> = d.metrics.iter().map(|m| m.key.as_str()).collect();
349 +
        assert_eq!(keys, ["median_frame_ms", "p99_frame_ms", "scenes"]);
350 +
        assert_eq!(d.metrics[1].fail_above, Some(20.0));
351 +
        assert_eq!((d.metrics[2].value, d.metrics[2].unit.as_str()), (12.0, ""));
352 +
        assert_eq!(d.context["budget_cycles"], 268800);
353 +
354 +
        let minimal = parse_declaration("cov", b"kind = \"coverage\"\nfile = \"lcov.info\"\n").unwrap();
355 +
        assert_eq!((minimal.title.as_str(), minimal.file.as_deref()), ("cov", Some("lcov.info")));
356 +
357 +
        let bad = |t: &str| parse_declaration("r", t.as_bytes()).unwrap_err();
358 +
        assert!(bad("title = \"x\"").contains("`kind` is required"));
359 +
        assert!(bad("kind = \"graphs\"").contains("not known"));
360 +
        assert!(bad("kind = \"custom\"\ntitel = \"x\"").contains("`titel` is not known"));
361 +
        assert!(bad("kind = \"custom\"\n[metrics]\nfps = \"fast\"").contains("metric fps"));
362 +
        assert!(bad("kind = \"custom\"\n[metrics]\nfps = { value = 1, better = \"more\" }").contains("better"));
363 +
        assert!(bad("kind = \"custom\"\n[metrics]\nfps = { unit = \"Hz\" }").contains("`value` is required"));
364 +
        assert!(bad("kind = \"custom\"\ntitle = ").starts_with("line 2"), "{}", bad("kind = \"custom\"\ntitle = "));
365 +
    }
366 +
367 +
    #[test]
368 +
    fn manifests_are_checked() {
369 +
        let blob = "a".repeat(64);
370 +
        let m = |files: &str| parse_manifest(format!("{{\"version\":1,\"files\":[{files}]}}").as_bytes());
371 +
        let file = |p: &str| format!("{{\"path\":\"{p}\",\"blob\":\"{blob}\",\"size\":3}}");
372 +
        let ok = m(&[file("report.toml"), file("data/a.csv")].join(",")).unwrap();
373 +
        assert_eq!((ok.files.len(), ok.bytes()), (2, 6));
374 +
        assert!(m(&file("index.html")).unwrap_err().contains("report.toml"));
375 +
        assert!(m(&[file("report.toml"), file("../x")].join(",")).is_err());
376 +
        assert!(m(&[file("report.toml"), file("a b")].join(",")).is_err());
377 +
        assert!(m(&[file("report.toml"), file("report.toml")].join(",")).unwrap_err().contains("twice"));
378 +
        assert!(m("{\"path\":\"report.toml\",\"blob\":\"zz\",\"size\":1}").unwrap_err().contains("blob"));
379 +
        assert!(parse_manifest(b"{\"version\":2,\"files\":[]}").unwrap_err().contains("version 2"));
380 +
        assert!(path_ok("assets/js/chart.min.js") && !path_ok("/abs") && !path_ok("a//b") && !path_ok(""));
381 +
    }
382 +
}
modified crates/hub/src/api/mod.rs +4 −0
…
49 unchanged lines
50 50
            "/flow/jobs/{id}/artifacts/{name}",
51 51
            post(crate::flow::artifact).layer(axum::extract::DefaultBodyLimit::max(crate::flow::MAX_ARTIFACT_BYTES + (1 << 20))),
52 52
        )
53 +
        .route(
54 +
            "/flow/jobs/{id}/reports/{name}/files",
55 +
            post(crate::flow::report_file).layer(axum::extract::DefaultBodyLimit::max(crate::flow::MAX_ARTIFACT_BYTES + (1 << 20))),
56 +
        )
53 57
        .route("/projects/{owner}", get(projects::list).post(projects::create))
54 58
        .route("/projects/{owner}/{number}", get(projects::board).patch(projects::edit))
55 59
        .route("/projects/{owner}/{number}/cards", get(projects::cards).post(projects::add_card))
…
2318 unchanged lines
modified crates/hub/src/db/flow.rs +1 −1
…
837 unchanged lines
838 838
    Ok(())
839 839
}
840 840
841 -
/// Job `job_id`'s artifacts, by name.
842 841
/// Every artifact blob a repository's runs keep, for garbage collection to
843 842
/// keep too (#126): the database is the only thing that names them.
844 843
pub async fn artifact_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
…
5 unchanged lines
850 849
    .await
851 850
}
852 851
852 +
/// Job `job_id`'s artifacts, by name.
853 853
pub async fn artifacts(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Artifact>> {
854 854
    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = ? ORDER BY name").bind(job_id).fetch_all(pool).await
855 855
}
…
27 unchanged lines
modified crates/hub/src/flow.rs +97 −2
…
939 unchanged lines
940 940
    let row =
941 941
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
942 942
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
943 +
    // A report's artifact is its manifest (#127): checked, and its files
944 +
    // found in the store, before it is kept.
945 +
    let report = match out.kind {
946 +
        flow_core::OutputKind::Report => Some(report_manifest(&app, &repo, &name, &body).await?),
947 +
        flow_core::OutputKind::Artifact => None,
948 +
    };
943 949
    let content = blake3::hash(&body).to_hex().to_string();
944 950
    let size = body.len() as i64;
945 -
    let report_bytes = (name == crate::coverage::ARTIFACT).then(|| body.clone());
946 -
    let junit_bytes = (name == crate::junit::ARTIFACT).then(|| body.clone());
951 +
    let artifact = report.is_none();
952 +
    let report_bytes = (artifact && name == crate::coverage::ARTIFACT).then(|| body.clone());
953 +
    let junit_bytes = (artifact && name == crate::junit::ARTIFACT).then(|| body.clone());
947 954
    let blob = app
948 955
        .repos
949 956
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
…
1 unchanged lines
951 958
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the artifact: {e}")))?;
952 959
    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };
953 960
    fdb::add_artifact(&app.db, &a).await?;
961 +
    if let Some((manifest, declaration)) = report {
962 +
        let origin = crate::reports::Origin {
963 +
            repo_id: row.id,
964 +
            run_id: run.id,
965 +
            run_number: run.number,
966 +
            job_id: job.id,
967 +
            job: &job.name,
968 +
            bookmark: &run.bookmark,
969 +
            state: &run.state,
970 +
        };
971 +
        let said = match declaration {
972 +
            Ok(d) => match crate::reports::keep(&app, &origin, &a.name, &d, &manifest, &a.blob).await {
973 +
                Ok(_) => {
974 +
                    let s = if d.metrics.len() == 1 { "" } else { "s" };
975 +
                    format!("report {}: {} \"{}\", {} metric{s}", a.name, d.kind, d.title, d.metrics.len())
976 +
                }
977 +
                Err(e) => format!("report {}: not indexed: {e}", a.name),
978 +
            },
979 +
            Err(why) => {
980 +
                let d = crate::reports::Declaration::unreadable(&a.name, &why);
981 +
                crate::reports::keep(&app, &origin, &a.name, &d, &manifest, &a.blob).await.ok();
982 +
                format!("report {}: report.toml: {why}", a.name)
983 +
            }
984 +
        };
985 +
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
986 +
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
987 +
    }
954 988
    // A coverage report (#113) is read now; what it says, or why it could
955 989
    // not be read, goes in the job's log. It never fails the upload.
956 990
    if let Some(bytes) = report_bytes {
…
12 unchanged lines
969 1003
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
970 1004
    }
971 1005
    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())
1006 +
}
1007 +
1008 +
/// Check report manifest `body` (#127) and that the repository holds every
1009 +
/// file it names; answers it with the report's declaration, or why that
1010 +
/// could not be read.
1011 +
async fn report_manifest(
1012 +
    app: &Shared,
1013 +
    repo: &RepoName,
1014 +
    name: &str,
1015 +
    body: &[u8],
1016 +
) -> Result<(crate::reports::Manifest, Result<crate::reports::Declaration, String>), ApiError> {
1017 +
    let manifest = crate::reports::parse_manifest(body).map_err(|e| ApiError(StatusCode::BAD_REQUEST, e))?;
1018 +
    let refs: Vec<(String, oxy_core::hash::ObjectRef)> =
1019 +
        manifest.files.iter().filter_map(|f| Some((f.path.clone(), oxy_core::hash::ObjectRef::from_hex(&f.blob)?))).collect();
1020 +
    let declaration = manifest.blob("report.toml").and_then(oxy_core::hash::ObjectRef::from_hex);
1021 +
    let found = app
1022 +
        .repos
1023 +
        .read(repo, move |r, _| {
1024 +
            if let Some((path, _)) = refs.iter().find(|(_, b)| !r.store().has(b)) {
1025 +
                return Ok(Err(path.clone()));
1026 +
            }
1027 +
            Ok(Ok(declaration.map(|b| r.read_blob(&b)).transpose()?))
1028 +
        })
1029 +
        .await
1030 +
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("read the report: {e}")))?;
1031 +
    let toml = found.map_err(|path| ApiError(StatusCode::BAD_REQUEST, format!("report file {path} was not uploaded; upload it first")))?;
1032 +
    let declaration = crate::reports::parse_declaration(name, &toml.unwrap_or_default());
1033 +
    Ok((manifest, declaration))
1034 +
}
1035 +
1036 +
/// `POST /api/v1/flow/jobs/{id}/reports/{name}/files`: one file of a report
1037 +
/// the job declares (#127), its path in `X-Flow-Report-Path`, kept as a
1038 +
/// blob of the run's repository. Answers `{"blob"}`; the manifest naming
1039 +
/// the files follows as the report's artifact.
1040 +
pub async fn report_file(
1041 +
    State(app): State<Shared>,
1042 +
    headers: HeaderMap,
1043 +
    Path((id, name)): Path<(i64, String)>,
1044 +
    body: axum::body::Bytes,
1045 +
) -> Result<Response, ApiError> {
1046 +
    runner_auth(&app, &headers).await?;
1047 +
    let job = running_job(&app, id).await?;
1048 +
    let spec: flow_core::Job =
1049 +
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
1050 +
    if !spec.outputs.iter().any(|o| o.name == name && o.kind == flow_core::OutputKind::Report) {
1051 +
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no report {name:?}", job.name)));
1052 +
    }
1053 +
    let path = headers.get("x-flow-report-path").and_then(|v| v.to_str().ok()).unwrap_or_default();
1054 +
    if !crate::reports::path_ok(path) {
1055 +
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("report file {path:?}: give its path in X-Flow-Report-Path")));
1056 +
    }
1057 +
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
1058 +
    let row =
1059 +
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
1060 +
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
1061 +
    let blob = app
1062 +
        .repos
1063 +
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
1064 +
        .await
1065 +
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the report file: {e}")))?;
1066 +
    Ok(Json(serde_json::json!({ "blob": blob.to_hex() })).into_response())
972 1067
}
973 1068
974 1069
/// Parse and keep a job's JUnit report: what it says, or why not.
…
549 unchanged lines
modified crates/hub/src/lib.rs +1 −0
…
23 unchanged lines
24 24
pub mod oidc;
25 25
pub mod protocol;
26 26
pub mod render;
27 +
pub mod reports;
27 28
pub mod repos;
28 29
pub mod secrets;
29 30
pub mod signin;
…
139 unchanged lines
modified crates/hub/src/maintenance.rs +12 −4
…
38 unchanged lines
39 39
            continue;
40 40
        }
41 41
        let label = slug.clone();
42 -
        // Flow's artifacts are named only in the database: kept as roots,
43 -
        // or gc would prune them after the grace period (#126). A listing
44 -
        // that fails skips this repository rather than risk them.
45 -
        let keep: Vec<oxy_core::hash::ObjectRef> = match db::flow::artifact_blobs(&app.db, row.id).await {
42 +
        // Flow's artifacts and reports' files are named only in the
43 +
        // database: kept as roots, or gc would prune them after the grace
44 +
        // period (#126, #127). A listing that fails skips this repository
45 +
        // rather than risk them.
46 +
        let listed = match db::flow::artifact_blobs(&app.db, row.id).await {
47 +
            Ok(mut blobs) => crate::reports::file_blobs(&app.db, row.id).await.map(|files| {
48 +
                blobs.extend(files);
49 +
                blobs
50 +
            }),
51 +
            Err(e) => Err(e),
52 +
        };
53 +
        let keep: Vec<oxy_core::hash::ObjectRef> = match listed {
46 54
            Ok(blobs) => blobs.iter().filter_map(|b| oxy_core::hash::ObjectRef::from_hex(b)).collect(),
47 55
            Err(e) => {
48 56
                tracing::warn!("gc: {slug}: listing its artifacts: {e}; skipped");
…
57 unchanged lines
modified crates/hub/tests/flow.rs +91 −0
…
951 unchanged lines
952 952
953 953
/// oxygenflow#37: a failing job keeps a `when "always"` artifact; its
954 954
/// receipt names none, and the run page says the file is outside it.
955 +
async fn upload_report_file(app: &Shared, job: i64, name: &str, path: &str, bytes: &[u8]) -> (StatusCode, String) {
956 +
    let req = Request::post(format!("/api/v1/flow/jobs/{job}/reports/{name}/files"))
957 +
        .header(header::AUTHORIZATION, "Bearer runner-secret")
958 +
        .header(header::CONTENT_TYPE, "application/octet-stream")
959 +
        .header("X-Flow-Report-Path", path)
960 +
        .body(Body::from(bytes.to_vec()))
961 +
        .unwrap();
962 +
    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();
963 +
    let status = res.status();
964 +
    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
965 +
}
966 +
967 +
/// #127: a report's files arrive one by one, then its manifest as the
968 +
/// report's artifact; the hub indexes the declaration, metrics and files,
969 +
/// and gc keeps every file.
970 +
#[tokio::test(flavor = "multi_thread")]
971 +
async fn reports_are_kept_file_by_file_and_indexed() {
972 +
    let h = hub().await;
973 +
    create(&h.app, "andy", "demo").await;
974 +
    let url = format!("{}/andy/demo", h.base);
975 +
    let work = tempfile::tempdir().unwrap();
976 +
    let c = client(&url, work.path());
977 +
    write(
978 +
        work.path(),
979 +
        ".flow/pipeline.kdl",
980 +
        &format!(
981 +
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"perf\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n        report \"frame-timing\" {{\n            path \"/src/out/report\"\n            when \"always\"\n        }}\n    }}\n}}\n"
982 +
        ),
983 +
    );
984 +
    c.snapshot("test").unwrap();
985 +
    c.commit("Measure frames", &[]).unwrap();
986 +
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
987 +
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
988 +
    let a = claim(&h.app).await.unwrap();
989 +
    let id = a["job_id"].as_i64().unwrap();
990 +
991 +
    let toml = b"kind = \"performance\"\ntitle = \"Frame timing\"\nsummary = \"Median 16.4 ms\"\n\n[metrics]\nmedian_ms = { value = 16.4, unit = \"ms\", better = \"lower\", fail_above = 20 }\n\n[context]\ntarget_fps = 30\n";
992 +
    let page = b"<p>frames</p>".to_vec();
993 +
    let data: Vec<u8> = (0..100_000u32).map(|i| (i % 241) as u8).collect();
994 +
    let files: [(&str, &[u8]); 3] = [("data/frames.csv", &data), ("index.html", &page), ("report.toml", toml)];
995 +
996 +
    assert_eq!(upload_report_file(&h.app, id, "tool", "report.toml", toml).await.0, StatusCode::BAD_REQUEST, "not a report");
997 +
    assert_eq!(upload_report_file(&h.app, id, "frame-timing", "../x", toml).await.0, StatusCode::BAD_REQUEST, "not a clean path");
998 +
    let mut listed = vec![];
999 +
    for (path, bytes) in files {
1000 +
        let (st, text) = upload_report_file(&h.app, id, "frame-timing", path, bytes).await;
1001 +
        assert_eq!(st, StatusCode::OK, "{text}");
1002 +
        let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
1003 +
        assert_eq!(blob, oxy_core::repo::Repo::blob_ref(bytes).to_hex());
1004 +
        listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
1005 +
    }
1006 +
    let mut missing = listed.clone();
1007 +
    missing.push(serde_json::json!({ "path": "assets/x.js", "blob": "ab".repeat(32), "size": 1 }));
1008 +
    let (st, text) =
1009 +
        upload(&h.app, id, "frame-timing", &serde_json::to_vec(&serde_json::json!({ "version": 1, "files": missing })).unwrap()).await;
1010 +
    assert_eq!(st, StatusCode::BAD_REQUEST, "a file never uploaded: {text}");
1011 +
    assert!(text.contains("assets/x.js"), "{text}");
1012 +
    let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed })).unwrap();
1013 +
    let (st, text) = upload(&h.app, id, "frame-timing", &manifest).await;
1014 +
    assert_eq!(st, StatusCode::OK, "{text}");
1015 +
1016 +
    let (kind, title, summary, files_n, bytes): (String, String, String, i64, i64) =
1017 +
        sqlx::query_as("SELECT kind, title, summary, files, bytes FROM reports WHERE job_id = ? AND name = 'frame-timing'")
1018 +
            .bind(id)
1019 +
            .fetch_one(&h.app.db)
1020 +
            .await
1021 +
            .unwrap();
1022 +
    assert_eq!((kind.as_str(), title.as_str(), summary.as_str()), ("performance", "Frame timing", "Median 16.4 ms"));
1023 +
    assert_eq!((files_n, bytes), (3, (data.len() + page.len() + toml.len()) as i64));
1024 +
    let rid: i64 = sqlx::query_scalar("SELECT id FROM reports WHERE job_id = ?").bind(id).fetch_one(&h.app.db).await.unwrap();
1025 +
    let m = oxygenhub::reports::metrics(&h.app.db, rid).await.unwrap();
1026 +
    assert_eq!((m[0].key.as_str(), m[0].value, m[0].fail_above), ("median_ms", 16.4, Some(20.0)));
1027 +
    let log = db::flow::log(&h.app.db, id, 0, 1000).await.unwrap();
1028 +
    assert!(log.iter().any(|r| r.text.contains("report frame-timing: performance \"Frame timing\", 1 metric")), "{log:?}");
1029 +
    let _ = run;
1030 +
1031 +
    // gc keeps every file: only the database names them.
1032 +
    {
1033 +
        let name = RepoName::parse("andy/demo").unwrap();
1034 +
        let mut hosted = oxy_core::repo::Repo::open(&h.app.repos.dir(&name)).unwrap();
1035 +
        hosted.config.set("core.gc-grace", "0").unwrap();
1036 +
        hosted.save_config().unwrap();
1037 +
    }
1038 +
    let done = oxygenhub::maintenance::gc_all(&h.app).await;
1039 +
    assert!(done.iter().all(|d| !d.failed), "{done:?}");
1040 +
    let hosted = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap();
1041 +
    for (path, bytes) in files {
1042 +
        assert_eq!(hosted.read_blob(&oxy_core::repo::Repo::blob_ref(bytes)).unwrap(), bytes, "{path} survives gc");
1043 +
    }
1044 +
}
1045 +
955 1046
#[tokio::test(flavor = "multi_thread")]
956 1047
async fn a_failing_job_keeps_an_artifact_outside_its_receipt() {
957 1048
    let h = hub().await;
…
1408 unchanged lines
replaced deps/oxygenflow +0 −0

No line changes.

modified docs/reports.md +13 −0
…
67 unchanged lines
68 68
The runner uploads the directory as one bundle. `when "always"` keeps a
69 69
failing job's report, as for artifacts.
70 70
71 +
The bundle travels file by file: each to
72 +
`POST /api/v1/flow/jobs/{job}/reports/{name}/files` with its path in
73 +
`X-Flow-Report-Path`, answered with its blob; then the manifest,
74 +
`{"version":1,"files":[{"path","blob","size"}]}`, as the job's artifact
75 +
under the report's name, which the receipt names. The hub refuses a
76 +
manifest naming a file it does not hold, reads `report.toml` and indexes
77 +
the report; a `report.toml` it cannot read is kept as a custom report whose
78 +
summary says why, so the job's log and the tab show what to fix. Paths are
79 +
`/`-separated names of letters, digits, `.`, `_`, `-` and `+`; a report
80 +
holds at most 2000 files and 512 MiB.
81 +
71 82
**From anywhere else:**
72 83
`oh report upload frame-timing ./report --bookmark develop` attaches it to
73 84
the bookmark's current state, or `--change ID` to a change's. The API is
…
16 unchanged lines
90 101
     measured state and heads, time, and the tree's blob.
91 102
   - `report_metrics`: one row per metric per report: key, value, unit,
92 103
     better, thresholds.
104 +
   - `report_files`: one row per file per report: path, blob, size; what
105 +
     garbage collection keeps and the frame serves.
93 106
94 107
   History, trends, the scrubber and checks read only this. Coverage's
95 108
   and Tests' existing tables become their kinds' details, keyed by
…
76 unchanged lines