reports.rs 91.7%

343 of 374 lines run, 71 of 120 functions · run #528 on main · run missed
//! Reports (oxygenhub#127, docs/reports.md): a directory a Flow job writes
//! with `report.toml` at its top. The runner uploads each file as a blob of
//! the run's repository, then a manifest naming them as the job's artifact
//! under the report's name. The hub checks the manifest, reads the
//! declaration and indexes the report, its metrics and its files, so the
//! Reports tab, its history and checks read only the database, and garbage
//! collection keeps every file a kept report names.

use crate::db::{now_ms, Pool};
use crate::Shared;
use serde::Deserialize;

/// Files in one report, and their total size; the runner keeps to these.
pub const MAX_FILES: usize = 2000;
pub const MAX_BYTES: u64 = 512 << 20;
/// Largest `report.toml` read.
const MAX_DECLARATION: usize = 64 << 10;
/// Metrics kept from one report.
const MAX_METRICS: usize = 200;
/// The kinds a report may declare.
pub const KINDS: [&str; 4] = ["coverage", "tests", "performance", "custom"];

/// A report file's path: `/`-separated names of letters, digits, `.`,
/// `_`, `-` and `+`, never `.` or `..`, so it travels in a header and a URL
/// as it is.
pub fn path_ok(p: &str) -> bool {
    p.len() <= 1024
        && p.split('/').count() <= 32
        && p.split('/').all(|n| {
            !n.is_empty() && n != "." && n != ".." && n.len() <= 255 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-+".contains(&b))
        })
}

/// A report's manifest, as the runner uploads it.
#[derive(Clone, Debug, Deserialize)]
pub struct Manifest {
    pub version: u32,
    pub files: Vec<ManifestFile>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct ManifestFile {
    pub path: String,
    pub blob: String,
    pub size: u64,
}

impl Manifest {
    pub fn bytes(&self) -> u64 {
        self.files.iter().map(|f| f.size).sum()
    }

    /// The blob of the file at `path`.
    pub fn blob(&self, path: &str) -> Option<&str> {
        self.files.iter().find(|f| f.path == path).map(|f| f.blob.as_str())
    }
}

/// Read and check a manifest: version 1, clean unique paths, blob names,
/// within the limits, with `report.toml`.
pub fn parse_manifest(bytes: &[u8]) -> Result<Manifest, String> {
    let m: Manifest = serde_json::from_slice(bytes).map_err(|e| format!("the report manifest is not valid: {e}"))?;
    if m.version != 1 {
        return Err(format!("report manifest version {} is not known; this hub reads version 1", m.version));
    }
    if m.files.len() > MAX_FILES {
        return Err(format!("the report holds {} files; at most {MAX_FILES}", m.files.len()));
    }
    if m.bytes() > MAX_BYTES {
        return Err(format!("the report holds {} MiB; at most {} MiB", m.bytes() >> 20, MAX_BYTES >> 20));
    }
    let mut seen = std::collections::HashSet::new();
    for f in &m.files {
        if !path_ok(&f.path) {
            return Err(format!("report file {:?}: names may hold letters, digits, '.', '_', '-' and '+'", f.path));
        }
        if !seen.insert(f.path.as_str()) {
            return Err(format!("report file {} is named twice", f.path));
        }
        if oxy_core::hash::ObjectRef::from_hex(&f.blob).is_none() {
            return Err(format!("report file {}: {:?} is not a blob name", f.path, f.blob));
        }
    }
    if m.blob("report.toml").is_none() {
        return Err("the report has no report.toml at its top".into());
    }
    Ok(m)
}

/// What `report.toml` declares.
#[derive(Clone, Debug, PartialEq)]
pub struct Declaration {
    pub kind: String,
    pub title: String,
    pub summary: String,
    /// The data file of a built-in kind (`lcov.info`, `junit.xml`).
    pub file: Option<String>,
    pub metrics: Vec<Metric>,
    /// The free-form `[context]` table, as JSON.
    pub context: serde_json::Value,
}

/// A number the hub tracks across runs.
#[derive(Clone, Debug, PartialEq, sqlx::FromRow)]
pub struct Metric {
    pub key: String,
    pub value: f64,
    pub unit: String,
    /// `higher`, `lower`, or empty when neither is better.
    pub better: String,
    pub fail_above: Option<f64>,
    pub fail_below: Option<f64>,
}

impl Declaration {
    /// The declaration of a report whose `report.toml` could not be read:
    /// kept as a custom report saying why, so its files stay and the reader
    /// sees what to fix.
    pub fn unreadable(name: &str, why: &str) -> Declaration {
        Declaration {
            kind: "custom".into(),
            title: name.into(),
            summary: format!("report.toml: {why}"),
            file: None,
            metrics: vec![],
            context: serde_json::json!({}),
        }
    }
}

fn number(v: &toml::Value) -> Option<f64> {
    match v {
        toml::Value::Integer(i) => Some(*i as f64),
        toml::Value::Float(f) if f.is_finite() => Some(*f),
        _ => None,
    }
}

fn metric_key_ok(k: &str) -> bool {
    !k.is_empty() && k.len() <= 64 && k.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b))
}

/// Read `report.toml` for report `name`; the error says what to fix.
pub fn parse_declaration(name: &str, bytes: &[u8]) -> Result<Declaration, String> {
    if bytes.len() > MAX_DECLARATION {
        return Err(format!("{} KiB; at most {} KiB", bytes.len() >> 10, MAX_DECLARATION >> 10));
    }
    let text = std::str::from_utf8(bytes).map_err(|_| "not UTF-8 text".to_string())?;
    let table: toml::Table = toml::from_str(text).map_err(|e| {
        let line = e.span().map(|s| text[..s.start.min(text.len())].lines().count().max(1));
        let what = if e.message().trim().is_empty() { "not valid TOML" } else { e.message().trim() };
        match line {
            Some(n) => format!("line {n}: {what}"),
            None => what.to_string(),
        }
    })?;
    let text_of = |key: &str, max: usize| -> Result<Option<String>, String> {
        match table.get(key) {
            None => Ok(None),
            Some(toml::Value::String(s)) if s.len() <= max => Ok(Some(s.clone())),
            Some(toml::Value::String(_)) => Err(format!("`{key}` is longer than {max} bytes")),
            Some(_) => Err(format!("`{key}` is a string")),
        }
    };
    for key in table.keys() {
        if !matches!(key.as_str(), "kind" | "title" | "summary" | "file" | "metrics" | "context") {
            return Err(format!("`{key}` is not known (kind, title, summary, file, metrics, context)"));
        }
    }
    let kind = text_of("kind", 32)?.ok_or("`kind` is required: one of coverage, tests, performance, custom")?;
    if !KINDS.contains(&kind.as_str()) {
        return Err(format!("kind {kind:?} is not known: one of coverage, tests, performance, custom"));
    }
    let file = text_of("file", 1024)?;
    if let Some(f) = &file {
        if !path_ok(f) {
            return Err(format!("file {f:?} is not a path in the report"));
        }
    }
    let mut metrics = Vec::new();
    match table.get("metrics") {
        None => {}
        Some(toml::Value::Table(t)) => {
            if t.len() > MAX_METRICS {
                return Err(format!("{} metrics; at most {MAX_METRICS}", t.len()));
            }
            for (key, v) in t {
                if !metric_key_ok(key) {
                    return Err(format!("metric {key:?}: names hold letters, digits, '.', '_' and '-'"));
                }
                metrics.push(parse_metric(key, v).map_err(|e| format!("metric {key}: {e}"))?);
            }
        }
        Some(_) => return Err("`metrics` is a table".into()),
    }
    let context = match table.get("context") {
        None => serde_json::json!({}),
        Some(v @ toml::Value::Table(_)) => serde_json::to_value(v).map_err(|e| format!("context: {e}"))?,
        Some(_) => return Err("`context` is a table".into()),
    };
    Ok(Declaration {
        kind,
        title: text_of("title", 200)?.unwrap_or_else(|| name.to_string()),
        summary: text_of("summary", 1000)?.unwrap_or_default(),
        file,
        metrics,
        context,
    })
}

fn parse_metric(key: &str, v: &toml::Value) -> Result<Metric, String> {
    let mut m = Metric { key: key.into(), value: 0.0, unit: String::new(), better: String::new(), fail_above: None, fail_below: None };
    if let Some(n) = number(v) {
        m.value = n;
        return Ok(m);
    }
    let toml::Value::Table(t) = v else {
        return Err("a number, or a table with `value`".into());
    };
    for (k, v) in t {
        match k.as_str() {
            "value" => m.value = number(v).ok_or("`value` is a number")?,
            "unit" => match v {
                toml::Value::String(s) if s.len() <= 32 => m.unit = s.clone(),
                _ => return Err("`unit` is a short string".into()),
            },
            "better" => match v.as_str() {
                Some(s @ ("higher" | "lower")) => m.better = s.into(),
                _ => return Err("`better` is \"higher\" or \"lower\"".into()),
            },
            "fail_above" => m.fail_above = Some(number(v).ok_or("`fail_above` is a number")?),
            "fail_below" => m.fail_below = Some(number(v).ok_or("`fail_below` is a number")?),
            _ => return Err(format!("`{k}` is not known (value, unit, better, fail_above, fail_below)")),
        }
    }
    if !t.contains_key("value") {
        return Err("`value` is required".into());
    }
    Ok(m)
}

/// Where a report came from.
pub struct Origin<'a> {
    pub repo_id: i64,
    pub run_id: i64,
    pub run_number: i64,
    pub job_id: i64,
    pub job: &'a str,
    pub bookmark: &'a str,
    pub state: &'a str,
}

/// Index report `name` of a job, replacing any the job kept before.
pub async fn keep(app: &Shared, o: &Origin<'_>, name: &str, d: &Declaration, m: &Manifest, manifest: &str) -> sqlx::Result<i64> {
    let mut tx = crate::db::begin_write(&app.db).await?;
    sqlx::query("DELETE FROM reports WHERE job_id = $1 AND name = $2").bind(o.job_id).bind(name).execute(&mut *tx).await?;
    let id: i64 = sqlx::query_scalar(
        "INSERT INTO reports (repo_id, name, kind, title, summary, context, run_id, job_id, run_number, job, bookmark, state, \
         manifest, files, bytes, created_ms) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) RETURNING id",
    )
    .bind(o.repo_id)
    .bind(name)
    .bind(&d.kind)
    .bind(&d.title)
    .bind(&d.summary)
    .bind(d.context.to_string())
    .bind(o.run_id)
    .bind(o.job_id)
    .bind(o.run_number)
    .bind(o.job)
    .bind(o.bookmark)
    .bind(o.state)
    .bind(manifest)
    .bind(m.files.len() as i64)
    .bind(m.bytes() as i64)
    .bind(now_ms())
    .fetch_one(&mut *tx)
    .await?;
    for x in &d.metrics {
        sqlx::query(
            "INSERT INTO report_metrics (report_id, key, value, unit, better, fail_above, fail_below) VALUES ($1, $2, $3, $4, $5, $6, $7)",
        )
        .bind(id)
        .bind(&x.key)
        .bind(x.value)
        .bind(&x.unit)
        .bind(&x.better)
        .bind(x.fail_above)
        .bind(x.fail_below)
        .execute(&mut *tx)
        .await?;
    }
    for f in &m.files {
        sqlx::query("INSERT INTO report_files (report_id, path, blob, size) VALUES ($1, $2, $3, $4)")
            .bind(id)
            .bind(&f.path)
            .bind(&f.blob)
            .bind(f.size as i64)
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    mark(app, o.repo_id);
    Ok(id)
}

static HAS: std::sync::Mutex<Option<std::collections::HashSet<String>>> = std::sync::Mutex::new(None);

/// Per hub (tests run several in one process) and repository.
fn key(app: &Shared, repo_id: i64) -> String {
    format!("{}#{repo_id}", app.repos.root().display())
}

/// Whether a repository has reports, for the Reports tab.
pub fn has_reports(app: &Shared, repo_id: i64) -> bool {
    HAS.lock().unwrap_or_else(|p| p.into_inner()).as_ref().is_some_and(|s| s.contains(&key(app, repo_id)))
}

fn mark(app: &Shared, repo_id: i64) {
    HAS.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id));
}

/// At start: which repositories have reports.
pub async fn load(app: &Shared) -> sqlx::Result<()> {
    let ids: Vec<i64> = sqlx::query_scalar("SELECT DISTINCT repo_id FROM reports").fetch_all(&app.db).await?;
    for id in ids {
        mark(app, id);
    }
    Ok(())
}

/// A kept report (one job's, or one upload's).
#[derive(Clone, Debug, sqlx::FromRow)]
pub struct Report {
    pub id: i64,
    pub repo_id: i64,
    pub name: String,
    pub kind: String,
    pub title: String,
    pub summary: String,
    /// The `[context]` table as JSON.
    pub context: String,
    pub run_id: Option<i64>,
    pub job_id: Option<i64>,
    pub run_number: Option<i64>,
    pub job: String,
    pub bookmark: String,
    pub state: String,
    pub manifest: Option<String>,
    pub files: i64,
    pub bytes: i64,
    pub created_ms: i64,
    /// When its files expired under retention (§3.3); its metrics stay.
    pub files_expired_ms: Option<i64>,
}

impl Report {
    /// The `[context]` table's entries, as text, in order.
    pub fn context_rows(&self) -> Vec<(String, String)> {
        let v: serde_json::Value = serde_json::from_str(&self.context).unwrap_or_default();
        v.as_object()
            .map(|m| m.iter().map(|(k, v)| (k.clone(), v.as_str().map(String::from).unwrap_or_else(|| v.to_string()))).collect())
            .unwrap_or_default()
    }
}

/// The newest report of each name in a repository, on `bookmark` when
/// given (else on any), newest first.
pub async fn latest(pool: &Pool, repo_id: i64, bookmark: Option<&str>) -> sqlx::Result<Vec<Report>> {
    sqlx::query_as(
        "SELECT * FROM reports r WHERE repo_id = $1 AND ($2 IS NULL OR bookmark = $3) \
         AND id = (SELECT id FROM reports s WHERE s.repo_id = r.repo_id AND s.name = r.name AND ($4 IS NULL OR s.bookmark = $5) \
                   ORDER BY created_ms DESC, id DESC LIMIT 1) \
         ORDER BY created_ms DESC",
    )
    .bind(repo_id)
    .bind(bookmark)
    .bind(bookmark)
    .bind(bookmark)
    .bind(bookmark)
    .fetch_all(pool)
    .await
}

/// Report `name`'s history on `bookmark`, newest first.
pub async fn history(pool: &Pool, repo_id: i64, name: &str, bookmark: &str, limit: i64) -> sqlx::Result<Vec<Report>> {
    sqlx::query_as("SELECT * FROM reports WHERE repo_id = $1 AND name = $2 AND bookmark = $3 ORDER BY created_ms DESC, id DESC LIMIT $4")
        .bind(repo_id)
        .bind(name)
        .bind(bookmark)
        .bind(limit)
        .fetch_all(pool)
        .await
}

/// The bookmarks report `name` has runs on, most recent first.
pub async fn bookmarks_of(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Vec<String>> {
    sqlx::query_scalar("SELECT bookmark FROM reports WHERE repo_id = $1 AND name = $2 GROUP BY bookmark ORDER BY MAX(created_ms) DESC")
        .bind(repo_id)
        .bind(name)
        .fetch_all(pool)
        .await
}

/// One report of a repository.
pub async fn get(pool: &Pool, repo_id: i64, id: i64) -> sqlx::Result<Option<Report>> {
    sqlx::query_as("SELECT * FROM reports WHERE repo_id = $1 AND id = $2").bind(repo_id).bind(id).fetch_optional(pool).await
}

/// A file of report `id`: its blob and size.
pub async fn file(pool: &Pool, id: i64, path: &str) -> sqlx::Result<Option<(String, i64)>> {
    sqlx::query_as("SELECT blob, size FROM report_files WHERE report_id = $1 AND path = $2").bind(id).bind(path).fetch_optional(pool).await
}

/// Report `id`'s files, by path, with their sizes.
pub async fn files_of(pool: &Pool, id: i64) -> sqlx::Result<Vec<(String, i64)>> {
    sqlx::query_as("SELECT path, size FROM report_files WHERE report_id = $1 ORDER BY path").bind(id).fetch_all(pool).await
}

/// The reports a job published, by name.
pub async fn of_job(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<Report>> {
    sqlx::query_as("SELECT * FROM reports WHERE job_id = $1 ORDER BY name").bind(job_id).fetch_all(pool).await
}

/// The capability a report's files are served under: they are read from a
/// sandboxed frame, which carries no session, so the link itself is what
/// lets it in, and only readers of the repository are given it.
pub fn file_mac(app: &Shared, id: i64) -> String {
    app.cookie_key.mac(format!("report-files:{id}").as_bytes())[..32].to_string()
}

/// Runs of a report on a bookmark other than the default whose files are
/// kept (docs/reports.md §3.3); older runs keep their metrics only.
pub const KEEP_OTHER_RUNS: i64 = 50;

/// Expire report files under retention (docs/reports.md §3.3): every run on
/// a repository's default bookmark keeps them; on any other bookmark (an
/// upload for a change has none) the newest [`KEEP_OTHER_RUNS`] runs of each
/// report do. An expired report keeps its metrics and summary and loses
/// its `report_files` rows, so garbage collection may prune the blobs.
/// Answers how many reports expired.
pub async fn expire(pool: &Pool) -> sqlx::Result<u64> {
    let mut tx = crate::db::begin_write(pool).await?;
    let ids: Vec<i64> = sqlx::query_scalar(
        "SELECT r.id FROM reports r JOIN repositories p ON p.id = r.repo_id \
         WHERE r.files_expired_ms IS NULL AND r.bookmark != p.default_bookmark \
         AND (SELECT COUNT(*) FROM reports s WHERE s.repo_id = r.repo_id AND s.name = r.name AND s.bookmark = r.bookmark \
              AND (s.created_ms > r.created_ms OR (s.created_ms = r.created_ms AND s.id > r.id))) >= $1",
    )
    .bind(KEEP_OTHER_RUNS)
    .fetch_all(&mut *tx)
    .await?;
    let now = now_ms();
    for id in &ids {
        sqlx::query("DELETE FROM report_files WHERE report_id = $1").bind(id).execute(&mut *tx).await?;
        sqlx::query("UPDATE reports SET files_expired_ms = $1 WHERE id = $2").bind(now).bind(id).execute(&mut *tx).await?;
    }
    tx.commit().await?;
    Ok(ids.len() as u64)
}

/// Report `id`'s metrics, by key.
pub async fn metrics(pool: &Pool, id: i64) -> sqlx::Result<Vec<Metric>> {
    sqlx::query_as("SELECT key, value, unit, better, fail_above, fail_below FROM report_metrics WHERE report_id = $1 ORDER BY key")
        .bind(id)
        .fetch_all(pool)
        .await
}

/// Every file blob a repository's kept reports name, for garbage collection
/// to keep too: the database is the only thing that names them.
pub async fn file_blobs(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
    sqlx::query_scalar("SELECT DISTINCT f.blob FROM report_files f JOIN reports r ON r.id = f.report_id WHERE r.repo_id = $1")
        .bind(repo_id)
        .fetch_all(pool)
        .await
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn declarations_are_read_with_their_metrics_and_context() {
        let d = parse_declaration(
            "frame-timing",
            br#"
kind    = "performance"
title   = "Frame timing"
summary = "Median frame 16.4 ms"

[metrics]
median_frame_ms = { value = 16.4, unit = "ms", better = "lower" }
p99_frame_ms    = { value = 18.1, unit = "ms", better = "lower", fail_above = 20 }
scenes          = 12

[context]
target_fps    = 30
budget_cycles = 268800
"#,
        )
        .unwrap();
        assert_eq!((d.kind.as_str(), d.title.as_str()), ("performance", "Frame timing"));
        let keys: Vec<&str> = d.metrics.iter().map(|m| m.key.as_str()).collect();
        assert_eq!(keys, ["median_frame_ms", "p99_frame_ms", "scenes"]);
        assert_eq!(d.metrics[1].fail_above, Some(20.0));
        assert_eq!((d.metrics[2].value, d.metrics[2].unit.as_str()), (12.0, ""));
        assert_eq!(d.context["budget_cycles"], 268800);

        let minimal = parse_declaration("cov", b"kind = \"coverage\"\nfile = \"lcov.info\"\n").unwrap();
        assert_eq!((minimal.title.as_str(), minimal.file.as_deref()), ("cov", Some("lcov.info")));

        let bad = |t: &str| parse_declaration("r", t.as_bytes()).unwrap_err();
        assert!(bad("title = \"x\"").contains("`kind` is required"));
        assert!(bad("kind = \"graphs\"").contains("not known"));
        assert!(bad("kind = \"custom\"\ntitel = \"x\"").contains("`titel` is not known"));
        assert!(bad("kind = \"custom\"\n[metrics]\nfps = \"fast\"").contains("metric fps"));
        assert!(bad("kind = \"custom\"\n[metrics]\nfps = { value = 1, better = \"more\" }").contains("better"));
        assert!(bad("kind = \"custom\"\n[metrics]\nfps = { unit = \"Hz\" }").contains("`value` is required"));
        assert!(bad("kind = \"custom\"\ntitle = ").starts_with("line 2"), "{}", bad("kind = \"custom\"\ntitle = "));
    }

    #[test]
    fn manifests_are_checked() {
        let blob = "a".repeat(64);
        let m = |files: &str| parse_manifest(format!("{{\"version\":1,\"files\":[{files}]}}").as_bytes());
        let file = |p: &str| format!("{{\"path\":\"{p}\",\"blob\":\"{blob}\",\"size\":3}}");
        let ok = m(&[file("report.toml"), file("data/a.csv")].join(",")).unwrap();
        assert_eq!((ok.files.len(), ok.bytes()), (2, 6));
        assert!(m(&file("index.html")).unwrap_err().contains("report.toml"));
        assert!(m(&[file("report.toml"), file("../x")].join(",")).is_err());
        assert!(m(&[file("report.toml"), file("a b")].join(",")).is_err());
        assert!(m(&[file("report.toml"), file("report.toml")].join(",")).unwrap_err().contains("twice"));
        assert!(m("{\"path\":\"report.toml\",\"blob\":\"zz\",\"size\":1}").unwrap_err().contains("blob"));
        assert!(parse_manifest(b"{\"version\":2,\"files\":[]}").unwrap_err().contains("version 2"));
        assert!(path_ok("assets/js/chart.min.js") && !path_ok("/abs") && !path_ok("a//b") && !path_ok(""));
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/hub/src/reports.rs 91.7%

Keyboard
