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, 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            let unreadable = Unreadable {
50                kind: kind.into(),
51                id,
52                reason: error.to_string(),
53            };
54            use std::io::Write;
55            let _ = writeln!(std::io::stderr().lock(), "{unreadable}");
56            None
57        }
58    }
59}
60
61pub(super) fn in_run(
62    db: &rusqlite::Connection,
63    run: &crate::types::RunId,
64) -> Result<Vec<Unreadable>, Error> {
65    Ok(in_runs(db, std::slice::from_ref(run))?
66        .remove(run)
67        .unwrap_or_default())
68}
69
70/// Existing-state audit for current-input status and the monitor, without rejecting a listing.
71pub async fn read_unreadable(state: &Path, repo: Option<&Path>) -> Result<Vec<Unreadable>, String> {
72    super::receipts::check_files(state)?;
73    if !state
74        .join(super::DATABASE)
75        .try_exists()
76        .map_err(|error| error.to_string())?
77    {
78        return Ok(Vec::new());
79    }
80    let connection = tokio_rusqlite::Connection::open_with_flags(
81        state.join(super::DATABASE),
82        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
83    )
84    .await
85    .map_err(|error| error.to_string())?;
86    let repo = repo.map(crate::platform::path_text);
87    let result = connection
88        .call(move |db| -> Result<_, Error> {
89            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
90            let transaction = db.transaction()?;
91            if !super::receipts::schema_initialized(&transaction)? {
92                return Ok(Vec::new());
93            }
94            let mut statement =
95                transaction.prepare("SELECT id FROM runs WHERE ?1 IS NULL OR repo=?1")?;
96            let runs = statement
97                .query_map(params![repo], |row| row.get::<_, crate::types::RunId>(0))?
98                .collect::<Result<Vec<_>, _>>()?;
99            let mut unreadable = Vec::new();
100            let mut seen = BTreeSet::new();
101            for error in in_runs(&transaction, &runs)?.into_values().flatten() {
102                if seen.insert((error.kind.clone(), error.id.clone())) {
103                    unreadable.push(error);
104                }
105            }
106            Ok(unreadable)
107        })
108        .await
109        .map_err(|error| error.to_string());
110    super::connection::finish(connection, result).await
111}
112
113/// Audit a bounded page in one query, including executions produced by it but no longer linked.
114pub(super) fn in_runs(
115    db: &rusqlite::Connection,
116    runs: &[crate::types::RunId],
117) -> Result<std::collections::BTreeMap<crate::types::RunId, Vec<Unreadable>>, Error> {
118    in_runs_cached(db, runs, &mut Audit::default())
119}
120
121#[derive(Default)]
122pub(super) struct Audit {
123    records: std::collections::BTreeMap<(String, String), (String, Option<Unreadable>)>,
124    pub selected: Option<crate::types::RunId>,
125    parsed: std::collections::BTreeMap<(String, String), Record>,
126}
127
128/// Only the selected Run's freshly decoded records survive its page audit; the selected loader
129/// consumes them, avoiding a second decode without caching every page's rich request bodies.
130enum Record {
131    Run(Box<super::Run>),
132    Request(Box<super::Request>),
133    Execution(Box<super::Execution>),
134}
135impl Record {
136    fn decode(kind: &str, id: &str, data: &str) -> Result<Self, Unreadable> {
137        match kind {
138            "run" => decode("run", id, data).map(|run| Self::Run(Box::new(run))),
139            "request" => {
140                decode("request", id, data).map(|request| Self::Request(Box::new(request)))
141            }
142            _ => {
143                decode("execution", id, data).map(|execution| Self::Execution(Box::new(execution)))
144            }
145        }
146    }
147}
148impl Audit {
149    fn take(&mut self, kind: &str, id: &str, data: &str) -> Option<Record> {
150        let key = (kind.to_owned(), id.to_owned());
151        (self
152            .records
153            .get(&key)
154            .is_some_and(|(saved, _)| saved == data))
155        .then(|| self.parsed.remove(&key))
156        .flatten()
157    }
158    pub fn take_run(&mut self, id: &crate::types::RunId, data: &str) -> Option<super::Run> {
159        match self.take("run", id, data)? {
160            Record::Run(run) => Some(*run),
161            _ => unreachable!("Run audit key"),
162        }
163    }
164    pub fn take_request(
165        &mut self,
166        id: &crate::types::RequestId,
167        data: &str,
168    ) -> Option<super::Request> {
169        match self.take("request", id, data)? {
170            Record::Request(request) => Some(*request),
171            _ => unreachable!("request audit key"),
172        }
173    }
174    pub fn take_execution(
175        &mut self,
176        id: &crate::types::ExecutionId,
177        data: &str,
178    ) -> Option<super::Execution> {
179        match self.take("execution", id, data)? {
180            Record::Execution(execution) => Some(*execution),
181            _ => unreachable!("execution audit key"),
182        }
183    }
184}
185
186pub(super) fn in_runs_cached(
187    db: &rusqlite::Connection,
188    runs: &[crate::types::RunId],
189    audit: &mut Audit,
190) -> Result<std::collections::BTreeMap<crate::types::RunId, Vec<Unreadable>>, Error> {
191    let mut statement = db.prepare_cached(
192        "WITH page(id) AS (SELECT value FROM json_each(?1))
193        SELECT r.id,'run',r.id,r.data FROM runs r JOIN page p ON p.id=r.id
194        UNION ALL SELECT q.run_id,'request',q.id,q.data FROM requests q JOIN page p ON p.id=q.run_id
195        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
196        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)",
197    )?;
198    let records = statement.query_map([serde_json::to_string(runs)?], |row| {
199        Ok((
200            row.get::<_, crate::types::RunId>(0)?,
201            row.get::<_, String>(1)?,
202            row.get::<_, String>(2)?,
203            row.get::<_, String>(3)?,
204        ))
205    })?;
206    audit.parsed.clear();
207    let mut unreadable: std::collections::BTreeMap<_, Vec<_>> = std::collections::BTreeMap::new();
208    let mut retained = std::collections::BTreeMap::new();
209    let mut seen = BTreeSet::new();
210    for record in records {
211        let (run, kind, id, data) = record?;
212        if !seen.insert((run.clone(), kind.clone(), id.clone())) {
213            continue;
214        }
215        let key = (kind.clone(), id.clone());
216        let error = if let Some((saved, error)) =
217            audit.records.get(&key).or_else(|| retained.get(&key))
218            && saved == &data
219        {
220            error.clone()
221        } else {
222            match Record::decode(&kind, &id, &data) {
223                Ok(record) => {
224                    if audit.selected.as_ref() == Some(&run) {
225                        audit.parsed.insert(key.clone(), record);
226                    }
227                    None
228                }
229                Err(error) => Some(error),
230            }
231        };
232        retained.insert(key, (data, error.clone()));
233        if let Some(error) = error {
234            unreadable.entry(run).or_default().push(error);
235        }
236    }
237    for records in unreadable.values_mut() {
238        records.sort_by(|a, b| (&a.kind, &a.id).cmp(&(&b.kind, &b.id)));
239    }
240    audit.records = retained;
241    Ok(unreadable)
242}
artifactize_core/store/unreadable.rs

artifactize_core/store/
unreadable.rs
