artifactize_core/store/
catalog.rs

1//! Global monitor catalog, intentionally independent of any visible Run page.
2use super::{
3    DATABASE,
4    receipts::{Error, check_files, schema_initialized},
5};
6use crate::{
7    repository::Identity,
8    types::{ExecutionId, RequestId, RequestStatus, RunStatus, Sha256Bytes},
9    workspace::{canonical_target, outside_workspaces},
10};
11use std::path::{Path, PathBuf};
12
13#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
14pub enum Signoff {
15    Execution(ExecutionId),
16    Request(RequestId),
17}
18
19#[derive(Debug, Clone)]
20pub struct CatalogRun {
21    pub repo_path: PathBuf,
22    pub repository: Identity,
23    pub status: RunStatus,
24    pub red: u64,
25    pub error: u64,
26    pub waiting: Vec<Signoff>,
27}
28
29/// Global attention facts; unreadable Runs are skipped individually with their reason.
30pub async fn read_catalog(state: &Path) -> Result<Vec<CatalogRun>, String> {
31    let state = canonical_target(state).map_err(|error| error.to_string())?;
32    check_files(&state)?;
33    if !state
34        .join(DATABASE)
35        .try_exists()
36        .map_err(|error| error.to_string())?
37    {
38        return Ok(Vec::new());
39    }
40    let connection = tokio_rusqlite::Connection::open_with_flags(
41        state.join(DATABASE),
42        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
43    )
44    .await
45    .map_err(|error| error.to_string())?;
46    let result = connection
47        .call(move |db| -> Result<_, Error> {
48            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
49            let transaction = db.transaction()?;
50            if !schema_initialized(&transaction)? {
51                return Ok(Vec::new());
52            }
53            let catalog = in_snapshot(&transaction, &state)?;
54            transaction.commit()?;
55            Ok(catalog)
56        })
57        .await
58        .map_err(|error| error.to_string());
59    super::connection::finish(connection, result).await
60}
61
62pub(super) fn in_snapshot(
63    db: &rusqlite::Connection,
64    state: &Path,
65) -> Result<Vec<CatalogRun>, Error> {
66    in_snapshot_cached(db, state, &mut Facts::default())
67}
68
69#[derive(Default)]
70pub(super) struct Facts {
71    runs: std::collections::BTreeMap<crate::types::RunId, (Sha256Bytes, Option<CatalogRun>)>,
72}
73
74pub(super) fn in_snapshot_cached(
75    db: &rusqlite::Connection,
76    state: &Path,
77    facts: &mut Facts,
78) -> Result<Vec<CatalogRun>, Error> {
79    let mut statement = db.prepare_cached("SELECT id,data FROM runs ORDER BY rowid")?;
80    let runs = statement
81        .query_map([], |row| {
82            Ok((
83                row.get::<_, crate::types::RunId>(0)?,
84                row.get::<_, String>(1)?,
85            ))
86        })?
87        .collect::<Result<Vec<_>, _>>()?;
88    let mut catalog = std::collections::BTreeMap::new();
89    let mut order = Vec::new();
90    let mut retained = std::collections::BTreeMap::new();
91    for (id, data) in runs {
92        let digest = Sha256Bytes::of(data.as_bytes());
93        let fact = match facts.runs.remove(&id) {
94            Some((saved, fact)) if saved == digest => fact,
95            _ => super::unreadable::evidence::<super::Run>("run", &data).map(|run| {
96                let status = run.status();
97                CatalogRun {
98                    repo_path: run.repo_path,
99                    repository: run.repository,
100                    status,
101                    red: 0,
102                    error: 0,
103                    waiting: Vec::new(),
104                }
105            }),
106        };
107        retained.insert(id.clone(), (digest, fact.clone()));
108        if let Some(fact) = fact {
109            order.push(id.clone());
110            catalog.insert(id, fact);
111        }
112    }
113    outside_workspaces(catalog.values().map(|run| run.repo_path.as_path()), state)
114        .map_err(|error| Error::Invalid(error.to_string()))?;
115    facts.runs = retained;
116    let mut statement = db.prepare_cached(
117        "SELECT q.run_id,q.status,q.execution_id,coalesce(json_extract(e.data,'$.provenance.requestId'),json_extract(q.data,'$.provenance.requestId'),q.id)
118        FROM requests q LEFT JOIN executions e ON e.id=q.execution_id WHERE q.status IN ('RED','ERROR','WAITING_HUMAN','DEFERRED') ORDER BY q.run_id,q.ordinal",
119    )?;
120    let attention = statement.query_map([], |row| {
121        Ok((
122            row.get::<_, crate::types::RunId>(0)?,
123            row.get::<_, RequestStatus>(1)?,
124            row.get::<_, Option<ExecutionId>>(2)?,
125            row.get::<_, RequestId>(3)?,
126        ))
127    })?;
128    for row in attention {
129        let (id, status, execution, request) = row?;
130        if let Some(run) = catalog.get_mut(&id) {
131            match status {
132                RequestStatus::Red => run.red += 1,
133                RequestStatus::Error => run.error += 1,
134                // A deferred sign-off waits for a person as well, after its Run has ended too.
135                RequestStatus::WaitingHuman | RequestStatus::Deferred => run
136                    .waiting
137                    .push(execution.map_or(Signoff::Request(request), Signoff::Execution)),
138                _ => unreachable!("attention query filters states"),
139            }
140        }
141    }
142    Ok(order
143        .into_iter()
144        .filter_map(|id| catalog.remove(&id))
145        .collect())
146}
artifactize_core/store/catalog.rs

artifactize_core/store/
catalog.rs
