artifactize_core/store/
history.rs

1//! The key history: every completed GREEN/RED execution of a reuse key whose history columns
2//! (`completed_at`, `bytes`, `last_used`) are set. The latest by completion time is the one
3//! reused; the cache GC and `cache rm` clear the columns and keep the execution.
4
5use std::path::Path;
6
7use rusqlite::{OpenFlags, OptionalExtension, params};
8use serde::Serialize;
9use tokio_rusqlite::Connection;
10
11use super::{
12    DATABASE, Execution,
13    receipts::{Error, NotifyingConnection, schema_initialized},
14};
15
16/// Bound retained cache history, including many tiny results; eviction preserves execution audit.
17pub const MAX_ENTRIES: i64 = 10_000;
18/// Bound cached JSON payload bytes independently of the record-count ceiling.
19pub const MAX_BYTES: i64 = 1024 * 1024 * 1024;
20/// Refuse one oversized reusable record rather than letting it monopolize cache storage.
21pub const MAX_ENTRY_BYTES: usize = 16 * 1024 * 1024;
22
23/// A record of the key history, of executions `e`.
24pub(super) const RECORD: &str =
25    "e.completed_at IS NOT NULL AND e.status IN ('GREEN','RED') AND e.key IS NOT NULL";
26
27/// The latest record of a key first: the newest completion, then the most recently stored.
28pub(super) const LATEST: &str = "e.completed_at DESC, e.rowid DESC";
29
30/// One record of a key's history, or with `cache list` the latest record of each key.
31#[derive(Debug, Serialize)]
32#[serde(rename_all = "camelCase")]
33pub struct Entry {
34    pub key: crate::types::ReuseKey,
35    pub eval_def_hash: crate::types::DefinitionHash,
36    pub execution_id: crate::types::ExecutionId,
37    pub verdict: crate::types::ExecutionStatus,
38    #[serde(with = "crate::platform::path_serde")]
39    pub repo_path: std::path::PathBuf,
40    pub eval_id: crate::types::EvalId,
41    /// The target Artifact's fingerprint.
42    pub fingerprint: Option<crate::types::Fingerprint>,
43    pub completed_at: Option<crate::types::Timestamp>,
44    /// `user@host` of the machine that produced the result.
45    pub producer: Option<String>,
46    /// The selected profile variant, when the result came from one.
47    pub variant: Option<String>,
48    /// The remote store URL of a mirrored record.
49    pub origin: Option<crate::types::StoreUrl>,
50    /// How many records the key holds.
51    pub records: i64,
52    pub bytes: i64,
53    pub last_used: crate::types::Timestamp,
54}
55
56async fn open(state: &Path, writable: bool) -> Result<Option<NotifyingConnection>, String> {
57    let state = crate::workspace::canonical_target(state).map_err(|e| e.to_string())?;
58    super::receipts::check_files(&state)?;
59    let path = state.join(DATABASE);
60    if !path.try_exists().map_err(|e| e.to_string())? {
61        return Ok(None);
62    }
63    let connection = Connection::open_with_flags(
64        path,
65        if writable {
66            OpenFlags::SQLITE_OPEN_READ_WRITE
67        } else {
68            OpenFlags::SQLITE_OPEN_READ_ONLY
69        },
70    )
71    .await
72    .map_err(|e| e.to_string())?;
73    let connection = NotifyingConnection::new(connection, &state);
74    let initialized = connection
75        .call(|db| -> Result<bool, Error> {
76            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
77            let transaction = db.transaction()?;
78            schema_initialized(&transaction)
79        })
80        .await
81        .map_err(|e| e.to_string())?;
82    Ok(initialized.then_some(connection))
83}
84
85/// The latest record of every key, or with `history` every record, latest first per key.
86pub async fn list(state: &Path, history: bool) -> Result<Vec<Entry>, String> {
87    let Some(connection) = open(state, false).await? else {
88        return Ok(Vec::new());
89    };
90    connection
91        .call(move |db| -> Result<_, Error> {
92            let mut statement = db.prepare(&format!(
93                "SELECT * FROM (SELECT e.key,e.eval_def_hash,e.id,e.status,json_extract(e.data,'$.provenance.repoPath'),json_extract(e.data,'$.provenance.evalId'),json_extract(e.data,'$.fingerprint'),json_extract(e.data,'$.completedAt'),json_extract(e.data,'$.producer.name'),json_extract(e.data,'$.options.variant'),json_extract(e.data,'$.origin.store'),count(*) OVER (PARTITION BY e.key),e.bytes,e.last_used,row_number() OVER (PARTITION BY e.key ORDER BY {LATEST}) AS rank,e.data FROM executions e WHERE {RECORD}) WHERE ?1 OR rank=1 ORDER BY 1,rank"
94            ))?;
95            Ok(statement
96                .query_map([history], |row| {
97                    let id: crate::types::ExecutionId = row.get(2)?;
98                    let data: String = row.get(15)?;
99                    if super::unreadable::evidence::<Execution>("execution", &data).is_none() {
100                        return Ok(None);
101                    }
102                    Ok(Some(Entry {
103                        key: row.get(0)?,
104                        eval_def_hash: row.get(1)?,
105                        execution_id: id,
106                        verdict: row.get(3)?,
107                        repo_path: row.get::<_, String>(4)?.into(),
108                        eval_id: row.get(5)?,
109                        fingerprint: row.get(6)?,
110                        completed_at: row.get(7)?,
111                        producer: row.get(8)?,
112                        variant: row.get(9)?,
113                        origin: row.get(10)?,
114                        records: row.get(11)?,
115                        bytes: row.get(12)?,
116                        last_used: row.get(13)?,
117                    }))
118                })?
119                .filter_map(|row| row.transpose())
120                .collect::<Result<_, _>>()?)
121        })
122        .await
123        .map_err(|e| e.to_string())
124}
125
126/// A key's records, latest first; without `history` only the latest.
127pub async fn show(state: &Path, key: &str, history: bool) -> Result<Vec<Execution>, String> {
128    let Some(connection) = open(state, false).await? else {
129        return Ok(Vec::new());
130    };
131    let Ok(key) = key.parse::<crate::types::ReuseKey>() else {
132        return Ok(Vec::new());
133    };
134    connection
135        .call(move |db| -> Result<_, Error> {
136            let mut statement = db.prepare(&format!(
137                "SELECT e.data FROM executions e WHERE e.key=? AND {RECORD} ORDER BY {LATEST} LIMIT ?"
138            ))?;
139            statement
140                .query_map(params![key, if history { -1 } else { 1 }], |row| {
141                    row.get::<_, String>(0)
142                })?
143                .filter_map(|row| match row {
144                    Ok(data) => super::unreadable::evidence("execution", &data).map(Ok),
145                    Err(error) => Some(Err(Error::Sql(error))),
146                })
147                .collect()
148        })
149        .await
150        .map_err(|e| e.to_string())
151}
152
153/// The latest locally produced record (never a mirror) of each key after `after`, in key order.
154pub async fn local(
155    state: &Path,
156    after: Option<crate::types::ReuseKey>,
157    limit: usize,
158) -> Result<Vec<Execution>, String> {
159    let Some(connection) = open(state, false).await? else {
160        return Ok(Vec::new());
161    };
162    connection
163        .call(move |db| -> Result<_, Error> {
164            let mut statement = db.prepare(&format!(
165                "SELECT data FROM (SELECT e.key,e.data,row_number() OVER (PARTITION BY e.key ORDER BY {LATEST}) AS rank FROM executions e WHERE {RECORD} AND json_extract(e.data,'$.origin') IS NULL AND e.key>?) WHERE rank=1 ORDER BY key LIMIT ?"
166            ))?;
167            statement
168                .query_map(
169                    params![
170                        after.as_ref().map_or("", crate::types::ReuseKey::as_str),
171                        limit as i64
172                    ],
173                    |row| row.get::<_, String>(0),
174                )?
175                .filter_map(|row| match row {
176                    Ok(data) => super::unreadable::evidence("execution", &data).map(Ok),
177                    Err(error) => Some(Err(Error::Sql(error))),
178                })
179                .collect()
180        })
181        .await
182        .map_err(|e| e.to_string())
183}
184
185/// Remove every record of an unused key from its history, preserving executions and Runs.
186pub async fn remove(state: &Path, key: &str) -> Result<bool, String> {
187    let Some(connection) = open(state, true).await? else {
188        return Ok(false);
189    };
190    let Ok(key) = key.parse::<crate::types::ReuseKey>() else {
191        return Ok(false);
192    };
193    connection
194        .call(move |db| -> Result<bool, Error> {
195            let transaction =
196                db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
197            let in_use: bool = transaction.query_row(
198                "SELECT EXISTS(SELECT 1 FROM executions WHERE key=?1 AND status IN ('RUNNING','WAITING_HUMAN')) OR EXISTS(SELECT 1 FROM requests q JOIN executions e ON e.id=q.execution_id WHERE e.key=?1 AND e.completed_at IS NOT NULL AND q.status IN ('QUEUED','RUNNING','WAITING_HUMAN'))",
199                [&key],
200                |row| row.get(0),
201            )?;
202            if in_use {
203                return Err(Error::Invalid(format!(
204                    "Key {key} is in use by an active execution or waiter."
205                )));
206            }
207            let removed = transaction.execute(
208                "UPDATE executions SET completed_at=NULL,bytes=NULL,last_used=NULL WHERE key=? AND completed_at IS NOT NULL",
209                [&key],
210            )?;
211            transaction.commit()?;
212            Ok(removed != 0)
213        })
214        .await
215        .map_err(|e| e.to_string())
216}
217
218/// The history columns of a completed record: its sortable completion time and size, or
219/// `None` when it does not join its key's history (no key, no verdict, or over the entry
220/// limit).
221pub(super) fn columns(
222    execution: &Execution,
223    bytes: usize,
224) -> Result<Option<(crate::types::Timestamp, i64)>, Error> {
225    let (Some(_), Some(_), Some(completed_at)) = (
226        &execution.key,
227        execution.verdict(),
228        &execution.completed_at(),
229    ) else {
230        return Ok(None);
231    };
232    if bytes > MAX_ENTRY_BYTES {
233        return Ok(None);
234    }
235    Ok(Some((*completed_at, bytes as i64)))
236}
237
238/// Mark a reused record as used now.
239pub(super) fn touch(
240    db: &rusqlite::Connection,
241    execution_id: &crate::types::ExecutionId,
242    at: crate::types::Timestamp,
243) -> Result<(), Error> {
244    db.execute(
245        "UPDATE executions SET last_used=? WHERE id=? AND completed_at IS NOT NULL",
246        params![at, execution_id],
247    )?;
248    Ok(())
249}
250
251/// Report a failed collection; the completed execution it followed is saved already.
252pub(super) fn collect_after(db: &mut rusqlite::Connection) {
253    if let Err(error) = collect(db) {
254        use std::io::Write;
255        let _ = writeln!(std::io::stderr().lock(), "Cache GC failed: {error}");
256    }
257}
258
259/// Drop least-recently-used records from the history while it is above its caps, and records
260/// over the entry limit, skipping keys with an active execution and records an in-flight
261/// request waits for.
262fn collect(db: &mut rusqlite::Connection) -> Result<(), Error> {
263    loop {
264        // Each eviction has its own short transaction; execution and receipt rows stay.
265        let transaction = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
266        let (entries, bytes): (i64, i64) = transaction.query_row(
267            "SELECT count(*),coalesce(sum(bytes),0) FROM executions WHERE completed_at IS NOT NULL",
268            [],
269            |row| Ok((row.get(0)?, row.get(1)?)),
270        )?;
271        let over_capacity = entries > MAX_ENTRIES || bytes > MAX_BYTES;
272        let candidate: Option<crate::types::ExecutionId> = transaction
273            .query_row(
274                "SELECT e.id FROM executions e WHERE e.completed_at IS NOT NULL AND (? OR e.bytes>?)
275             AND NOT EXISTS(SELECT 1 FROM executions a WHERE a.key=e.key AND a.status IN ('RUNNING','WAITING_HUMAN'))
276             AND NOT EXISTS(SELECT 1 FROM requests q WHERE q.execution_id=e.id AND q.status IN ('QUEUED','RUNNING','WAITING_HUMAN'))
277             ORDER BY e.last_used,e.key,e.completed_at LIMIT 1",
278                params![over_capacity, MAX_ENTRY_BYTES as i64],
279                |row| row.get(0),
280            )
281            .optional()?;
282        let Some(execution_id) = candidate else {
283            return Ok(transaction.commit()?);
284        };
285        transaction.execute(
286            "UPDATE executions SET completed_at=NULL,bytes=NULL,last_used=NULL WHERE id=?",
287            [execution_id],
288        )?;
289        transaction.commit()?;
290    }
291}
artifactize_core/store/history.rs

artifactize_core/store/
history.rs
