artifactize_core/store/
catalog.rs

1//! Global monitor catalog, intentionally independent of any visible Run page.
2use super::{
3    DATABASE, Unreadable,
4    receipts::{Error, check_files, schema_initialized},
5    unreadable::decode,
6};
7use crate::{
8    repository::Identity,
9    types::{ExecutionId, RequestId, RequestStatus, RunStatus},
10    workspace::{canonical_target, outside_workspaces},
11};
12use std::path::{Path, PathBuf};
13
14#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
15pub enum Signoff {
16    Execution(ExecutionId),
17    Request(RequestId),
18}
19
20#[derive(Debug, Clone)]
21pub struct CatalogRun {
22    pub id: crate::types::RunId,
23    pub created_at: crate::types::Timestamp,
24    pub completed_at: Option<crate::types::Timestamp>,
25    pub repo_path: PathBuf,
26    pub repository: Identity,
27    pub status: RunStatus,
28    pub red: u64,
29    pub error: u64,
30    pub waiting: Vec<Signoff>,
31}
32
33/// Global attention facts; unreadable Runs are skipped individually with their reason.
34pub async fn read_catalog(state: &Path) -> Result<Vec<CatalogRun>, String> {
35    let state = canonical_target(state).map_err(|error| error.to_string())?;
36    check_files(&state)?;
37    if !state
38        .join(DATABASE)
39        .try_exists()
40        .map_err(|error| error.to_string())?
41    {
42        return Ok(Vec::new());
43    }
44    let connection = tokio_rusqlite::Connection::open_with_flags(
45        state.join(DATABASE),
46        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
47    )
48    .await
49    .map_err(|error| error.to_string())?;
50    let result = connection
51        .call(move |db| -> Result<_, Error> {
52            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
53            let transaction = db.transaction()?;
54            if !schema_initialized(&transaction)? {
55                return Ok(Vec::new());
56            }
57            let catalog = in_snapshot(&transaction, &state, None)?;
58            transaction.commit()?;
59            Ok(catalog.runs)
60        })
61        .await
62        .map_err(|error| error.to_string());
63    super::connection::finish(connection, result).await
64}
65
66/// The saved Runs, oldest first, and the Run records that could not be read.
67pub(crate) struct Catalog {
68    pub runs: Vec<CatalogRun>,
69    pub unreadable: Vec<Unreadable>,
70}
71
72/// Read every saved Run's state, or only the state of the Runs `only` names, never their
73/// definitions: they live in their own row, so a Run's state is small however many Artifacts
74/// it has.
75pub(super) fn in_snapshot(
76    db: &rusqlite::Connection,
77    state: &Path,
78    only: Option<&[crate::types::RunId]>,
79) -> Result<Catalog, Error> {
80    let only = only.map(serde_json::to_string).transpose()?;
81    let mut statement = db.prepare_cached(
82        "SELECT id,data FROM runs WHERE ?1 IS NULL OR id IN (SELECT value FROM json_each(?1)) ORDER BY rowid",
83    )?;
84    let rows = statement
85        .query_map([&only], |row| {
86            Ok((
87                row.get::<_, crate::types::RunId>(0)?,
88                row.get::<_, String>(1)?,
89            ))
90        })?
91        .collect::<Result<Vec<_>, _>>()?;
92    let mut catalog = std::collections::BTreeMap::new();
93    let mut order = Vec::new();
94    let mut unreadable = Vec::new();
95    for (id, data) in rows {
96        match decode::<super::Run>("run", &id, &data) {
97            Ok(run) => {
98                let status = run.status();
99                order.push(id.clone());
100                catalog.insert(
101                    id,
102                    CatalogRun {
103                        id: run.id.clone(),
104                        created_at: run.created_at,
105                        completed_at: run.completed_at(),
106                        repo_path: run.repo_path,
107                        repository: run.repository,
108                        status,
109                        red: 0,
110                        error: 0,
111                        waiting: Vec::new(),
112                    },
113                );
114            }
115            Err(error) => unreadable.push(error),
116        }
117    }
118    outside_workspaces(catalog.values().map(|run| run.repo_path.as_path()), state)
119        .map_err(|error| Error::Invalid(error.to_string()))?;
120    // Counts come from the status column alone; only a sign-off still waiting reads its records.
121    let mut statement = db.prepare_cached(
122        "SELECT run_id,status,count(*) FROM requests WHERE status IN ('RED','ERROR')
123        AND (?1 IS NULL OR run_id IN (SELECT value FROM json_each(?1))) GROUP BY run_id,status",
124    )?;
125    let counts = statement.query_map([&only], |row| {
126        Ok((
127            row.get::<_, crate::types::RunId>(0)?,
128            row.get::<_, RequestStatus>(1)?,
129            row.get::<_, i64>(2)?,
130        ))
131    })?;
132    for row in counts {
133        let (id, status, count) = row?;
134        let count = u64::try_from(count).expect("COUNT(*) is nonnegative");
135        if let Some(run) = catalog.get_mut(&id) {
136            match status {
137                RequestStatus::Red => run.red = count,
138                RequestStatus::Error => run.error = count,
139                _ => unreachable!("count query filters states"),
140            }
141        }
142    }
143    let mut statement = db.prepare_cached(
144        "SELECT q.run_id,q.execution_id,coalesce(json_extract(e.data,'$.provenance.requestId'),json_extract(q.data,'$.provenance.requestId'),q.id)
145        FROM requests q LEFT JOIN executions e ON e.id=q.execution_id WHERE q.status IN ('WAITING_HUMAN','DEFERRED')
146        AND (?1 IS NULL OR q.run_id IN (SELECT value FROM json_each(?1))) ORDER BY q.run_id,q.ordinal",
147    )?;
148    let waiting = statement.query_map([&only], |row| {
149        Ok((
150            row.get::<_, crate::types::RunId>(0)?,
151            row.get::<_, Option<ExecutionId>>(1)?,
152            row.get::<_, RequestId>(2)?,
153        ))
154    })?;
155    for row in waiting {
156        let (id, execution, request) = row?;
157        // A deferred sign-off waits for a person as well, after its Run has ended too.
158        if let Some(run) = catalog.get_mut(&id) {
159            run.waiting
160                .push(execution.map_or(Signoff::Request(request), Signoff::Execution));
161        }
162    }
163    Ok(Catalog {
164        runs: order
165            .into_iter()
166            .filter_map(|id| catalog.remove(&id))
167            .collect(),
168        unreadable,
169    })
170}
artifactize_core/store/catalog.rs

artifactize_core/store/
catalog.rs
