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

artifactize_core/store/
catalog.rs
