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<crate::types::ProducerId>,
46    /// The selected profile variant, when the result came from one.
47    pub variant: Option<crate::config::ProfileVariantName>,
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    match initialized {
83        Ok(true) => Ok(Some(connection)),
84        result => connection.finish(result.map(|_| None)).await,
85    }
86}
87
88/// The latest record of every key, or with `history` every record, latest first per key.
89pub async fn list(state: &Path, history: bool) -> Result<Vec<Entry>, String> {
90    let Some(connection) = open(state, false).await? else {
91        return Ok(Vec::new());
92    };
93    let result = connection
94        .call(move |db| -> Result<_, Error> {
95            let mut statement = db.prepare(&format!(
96                "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,'$.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"
97            ))?;
98            Ok(statement
99                .query_map([history], |row| {
100                    let id: crate::types::ExecutionId = row.get(2)?;
101                    let data: String = row.get(13)?;
102                    let Some(execution) =
103                        super::unreadable::evidence::<Execution>("execution", &data)
104                    else {
105                        return Ok(None);
106                    };
107                    Ok(Some(Entry {
108                        key: row.get(0)?,
109                        eval_def_hash: row.get(1)?,
110                        execution_id: id,
111                        verdict: row.get(3)?,
112                        repo_path: row.get::<_, String>(4)?.into(),
113                        eval_id: row.get(5)?,
114                        fingerprint: row.get(6)?,
115                        completed_at: row.get(7)?,
116                        producer: execution.producer.map(|producer| producer.name),
117                        variant: execution.options.variant,
118                        origin: row.get(8)?,
119                        records: row.get(9)?,
120                        bytes: row.get(10)?,
121                        last_used: row.get(11)?,
122                    }))
123                })?
124                .filter_map(|row| row.transpose())
125                .collect::<Result<_, _>>()?)
126        })
127        .await
128        .map_err(|e| e.to_string());
129    connection.finish(result).await
130}
131
132/// A key's records, latest first; without `history` only the latest.
133pub async fn show(state: &Path, key: &str, history: bool) -> Result<Vec<Execution>, String> {
134    let Some(connection) = open(state, false).await? else {
135        return Ok(Vec::new());
136    };
137    let Ok(key) = key.parse::<crate::types::ReuseKey>() else {
138        return connection.finish(Ok(Vec::new())).await;
139    };
140    let result = connection
141        .call(move |db| -> Result<_, Error> {
142            let mut statement = db.prepare(&format!(
143                "SELECT e.data FROM executions e WHERE e.key=? AND {RECORD} ORDER BY {LATEST} LIMIT ?"
144            ))?;
145            statement
146                .query_map(params![key, if history { -1 } else { 1 }], |row| {
147                    row.get::<_, String>(0)
148                })?
149                .filter_map(|row| match row {
150                    Ok(data) => super::unreadable::evidence("execution", &data).map(Ok),
151                    Err(error) => Some(Err(Error::Sql(error))),
152                })
153                .collect()
154        })
155        .await
156        .map_err(|e| e.to_string());
157    connection.finish(result).await
158}
159
160/// The latest locally produced record (never a mirror) of each key after `after`, in key order.
161pub async fn local(
162    state: &Path,
163    after: Option<crate::types::ReuseKey>,
164    limit: usize,
165) -> Result<Vec<Execution>, String> {
166    let Some(connection) = open(state, false).await? else {
167        return Ok(Vec::new());
168    };
169    let result = connection
170        .call(move |db| -> Result<_, Error> {
171            let mut statement = db.prepare(&format!(
172                "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 ?"
173            ))?;
174            statement
175                .query_map(
176                    params![
177                        after.as_ref().map_or("", crate::types::ReuseKey::as_str),
178                        limit as i64
179                    ],
180                    |row| row.get::<_, String>(0),
181                )?
182                .filter_map(|row| match row {
183                    Ok(data) => super::unreadable::evidence("execution", &data).map(Ok),
184                    Err(error) => Some(Err(Error::Sql(error))),
185                })
186                .collect()
187        })
188        .await
189        .map_err(|e| e.to_string());
190    connection.finish(result).await
191}
192
193/// Remove every record of an unused key from its history, preserving executions and Runs.
194pub async fn remove(state: &Path, key: &str) -> Result<bool, String> {
195    let Some(connection) = open(state, true).await? else {
196        return Ok(false);
197    };
198    let Ok(key) = key.parse::<crate::types::ReuseKey>() else {
199        return connection.finish(Ok(false)).await;
200    };
201    let result = connection
202        .call(move |db| -> Result<bool, Error> {
203            let transaction =
204                db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
205            let in_use: bool = transaction.query_row(
206                "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'))",
207                [&key],
208                |row| row.get(0),
209            )?;
210            if in_use {
211                return Err(Error::Invalid(format!(
212                    "Key {key} is in use by an active execution or waiter."
213                )));
214            }
215            let removed = transaction.execute(
216                "UPDATE executions SET completed_at=NULL,bytes=NULL,last_used=NULL WHERE key=? AND completed_at IS NOT NULL",
217                [&key],
218            )?;
219            transaction.commit()?;
220            Ok(removed != 0)
221        })
222        .await
223        .map_err(|e| e.to_string());
224    connection.finish(result).await
225}
226
227/// The history columns of a completed record: its sortable completion time and size, or
228/// `None` when it does not join its key's history (no key, no verdict, or over the entry
229/// limit).
230pub(super) fn columns(
231    execution: &Execution,
232    bytes: usize,
233) -> Result<Option<(crate::types::Timestamp, i64)>, Error> {
234    let (Some(_), Some(_), Some(completed_at)) = (
235        &execution.key,
236        execution.verdict(),
237        &execution.completed_at(),
238    ) else {
239        return Ok(None);
240    };
241    if bytes > MAX_ENTRY_BYTES {
242        return Ok(None);
243    }
244    Ok(Some((*completed_at, bytes as i64)))
245}
246
247/// Mark a reused record as used now.
248pub(super) fn touch(
249    db: &rusqlite::Connection,
250    execution_id: &crate::types::ExecutionId,
251    at: crate::types::Timestamp,
252) -> Result<(), Error> {
253    db.execute(
254        "UPDATE executions SET last_used=? WHERE id=? AND completed_at IS NOT NULL",
255        params![at, execution_id],
256    )?;
257    Ok(())
258}
259
260/// Report a failed collection; the completed execution it followed is saved already.
261pub(super) fn collect_after(db: &mut rusqlite::Connection) {
262    if let Err(error) = collect(db) {
263        use std::io::Write;
264        let _ = writeln!(std::io::stderr().lock(), "Cache GC failed: {error}");
265    }
266}
267
268/// Drop least-recently-used records from the history while it is above its caps, and records
269/// over the entry limit, skipping keys with an active execution and records an in-flight
270/// request waits for.
271fn collect(db: &mut rusqlite::Connection) -> Result<(), Error> {
272    loop {
273        // Each eviction has its own short transaction; execution and receipt rows stay.
274        let transaction = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
275        let (entries, bytes): (i64, i64) = transaction.query_row(
276            "SELECT count(*),coalesce(sum(bytes),0) FROM executions WHERE completed_at IS NOT NULL",
277            [],
278            |row| Ok((row.get(0)?, row.get(1)?)),
279        )?;
280        let over_capacity = entries > MAX_ENTRIES || bytes > MAX_BYTES;
281        let candidate: Option<crate::types::ExecutionId> = transaction
282            .query_row(
283                "SELECT e.id FROM executions e WHERE e.completed_at IS NOT NULL AND (? OR e.bytes>?)
284             AND NOT EXISTS(SELECT 1 FROM executions a WHERE a.key=e.key AND a.status IN ('RUNNING','WAITING_HUMAN'))
285             AND NOT EXISTS(SELECT 1 FROM requests q WHERE q.execution_id=e.id AND q.status IN ('QUEUED','RUNNING','WAITING_HUMAN'))
286             ORDER BY e.last_used,e.key,e.completed_at LIMIT 1",
287                params![over_capacity, MAX_ENTRY_BYTES as i64],
288                |row| row.get(0),
289            )
290            .optional()?;
291        let Some(execution_id) = candidate else {
292            return Ok(transaction.commit()?);
293        };
294        transaction.execute(
295            "UPDATE executions SET completed_at=NULL,bytes=NULL,last_used=NULL WHERE id=?",
296            [execution_id],
297        )?;
298        transaction.commit()?;
299    }
300}
artifactize_core/store/history.rs

artifactize_core/store/
history.rs
