artifactize_core/store/
unreadable.rs

1//! Per-record corruption reporting, separate from valid lifecycle models.
2
3use std::{collections::BTreeSet, path::Path};
4
5use rusqlite::params;
6use serde::{Deserialize, Serialize, de::DeserializeOwned};
7
8use super::receipts::Error;
9
10#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
11#[serde(rename_all = "camelCase")]
12pub struct Unreadable {
13    pub kind: String,
14    pub id: String,
15    pub reason: String,
16}
17
18impl std::fmt::Display for Unreadable {
19    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20        write!(out, "Unreadable {} {}: {}", self.kind, self.id, self.reason)
21    }
22}
23
24pub(super) fn decode<T: DeserializeOwned>(
25    kind: &'static str,
26    id: &str,
27    data: &str,
28) -> Result<T, Unreadable> {
29    serde_json::from_str(data).map_err(|error| Unreadable {
30        kind: kind.into(),
31        id: id.to_owned(),
32        reason: error.to_string(),
33    })
34}
35
36/// Reads used only for evidence skip one bad record and explain which record was skipped.
37pub(super) fn evidence<T: DeserializeOwned>(kind: &'static str, data: &str) -> Option<T> {
38    match serde_json::from_str(data) {
39        Ok(record) => Some(record),
40        Err(error) => {
41            // Decode the identity only on failure, not the whole payload a second time.
42            #[derive(Deserialize)]
43            struct Identity {
44                id: String,
45            }
46            let id = serde_json::from_str::<Identity>(data)
47                .map(|identity| identity.id)
48                .unwrap_or_else(|_| "(unknown id)".into());
49            skipped(&Unreadable {
50                kind: kind.into(),
51                id,
52                reason: error.to_string(),
53            });
54            None
55        }
56    }
57}
58
59/// Explain a record a read skipped.
60pub(super) fn skipped(unreadable: &Unreadable) {
61    use std::io::Write;
62    let _ = writeln!(std::io::stderr().lock(), "{unreadable}");
63}
64
65pub(super) fn in_run(
66    db: &rusqlite::Connection,
67    run: &crate::types::RunId,
68) -> Result<Vec<Unreadable>, Error> {
69    Ok(in_runs(db, std::slice::from_ref(run))?
70        .remove(run)
71        .unwrap_or_default())
72}
73
74/// Existing-state audit for current-input status and the monitor, without rejecting a listing.
75pub async fn read_unreadable(state: &Path, repo: Option<&Path>) -> Result<Vec<Unreadable>, String> {
76    super::receipts::check_files(state)?;
77    if !state
78        .join(super::DATABASE)
79        .try_exists()
80        .map_err(|error| error.to_string())?
81    {
82        return Ok(Vec::new());
83    }
84    let connection = tokio_rusqlite::Connection::open_with_flags(
85        state.join(super::DATABASE),
86        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
87    )
88    .await
89    .map_err(|error| error.to_string())?;
90    let repo = repo.map(crate::platform::path_text);
91    let result = connection
92        .call(move |db| -> Result<_, Error> {
93            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
94            let transaction = db.transaction()?;
95            if !super::receipts::schema_initialized(&transaction)? {
96                return Ok(Vec::new());
97            }
98            let mut statement =
99                transaction.prepare("SELECT id FROM runs WHERE ?1 IS NULL OR repo=?1")?;
100            let runs = statement
101                .query_map(params![repo], |row| row.get::<_, crate::types::RunId>(0))?
102                .collect::<Result<Vec<_>, _>>()?;
103            let mut unreadable = Vec::new();
104            let mut seen = BTreeSet::new();
105            for error in in_runs(&transaction, &runs)?.into_values().flatten() {
106                if seen.insert((error.kind.clone(), error.id.clone())) {
107                    unreadable.push(error);
108                }
109            }
110            Ok(unreadable)
111        })
112        .await
113        .map_err(|error| error.to_string());
114    super::connection::finish(connection, result).await
115}
116
117/// Audit the given Runs in one query: their state and definitions rows, their requests and
118/// executions, including executions produced by them but no longer linked. Every record is
119/// decoded once; nothing is kept between audits.
120pub(super) fn in_runs(
121    db: &rusqlite::Connection,
122    runs: &[crate::types::RunId],
123) -> Result<std::collections::BTreeMap<crate::types::RunId, Vec<Unreadable>>, Error> {
124    let mut statement = db.prepare_cached(
125        "WITH page(id) AS (SELECT value FROM json_each(?1))
126        SELECT r.id,'run',r.id,r.data FROM runs r JOIN page p ON p.id=r.id
127        UNION ALL SELECT d.run_id,'definitions',d.run_id,d.data FROM run_definitions d JOIN page p ON p.id=d.run_id
128        UNION ALL SELECT q.run_id,'request',q.id,q.data FROM requests q JOIN page p ON p.id=q.run_id
129        UNION ALL SELECT q.run_id,'execution',e.id,e.data FROM (SELECT DISTINCT run_id,execution_id FROM requests WHERE run_id IN (SELECT id FROM page) AND execution_id IS NOT NULL) q JOIN executions e ON e.id=q.execution_id
130        UNION ALL SELECT json_extract(e.data,'$.provenance.runId'),'execution',e.id,e.data FROM executions e WHERE json_extract(e.data,'$.provenance.runId') IN (SELECT id FROM page)",
131    )?;
132    let records = statement.query_map([serde_json::to_string(runs)?], |row| {
133        Ok((
134            row.get::<_, crate::types::RunId>(0)?,
135            row.get::<_, String>(1)?,
136            row.get::<_, String>(2)?,
137            row.get::<_, String>(3)?,
138        ))
139    })?;
140    let mut unreadable: std::collections::BTreeMap<_, Vec<_>> = std::collections::BTreeMap::new();
141    let mut seen = BTreeSet::new();
142    for record in records {
143        let (run, kind, id, data) = record?;
144        if !seen.insert((run.clone(), kind.clone(), id.clone())) {
145            continue;
146        }
147        let error = match kind.as_str() {
148            "run" => decode::<super::Run>("run", &id, &data).err(),
149            // A Run whose definitions do not read is an unreadable Run.
150            "definitions" => decode::<super::definitions::Definitions>("run", &id, &data).err(),
151            "request" => decode::<super::Request>("request", &id, &data).err(),
152            _ => decode::<super::Execution>("execution", &id, &data).err(),
153        };
154        if let Some(error) = error {
155            unreadable.entry(run).or_default().push(error);
156        }
157    }
158    for records in unreadable.values_mut() {
159        records.sort_by(|a, b| (&a.kind, &a.id).cmp(&(&b.kind, &b.id)));
160    }
161    Ok(unreadable)
162}
artifactize_core/store/unreadable.rs

artifactize_core/store/
unreadable.rs
