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, params};
8use serde::Serialize;
9use tokio_rusqlite::Connection;
10
11use super::{
12    DATABASE, Execution,
13    receipts::{Error, NotifyingConnection, schema_initialized},
14};
15
16/// Bound the history no checkout reuses now: records a newer one of the same eval and checkout
17/// replaced, and the records of checkouts that are gone. Eviction preserves execution audit.
18pub const MAX_ENTRIES: i64 = 10_000;
19/// Bound those records' JSON payload bytes independently of the record-count ceiling.
20pub const MAX_BYTES: i64 = 1024 * 1024 * 1024;
21/// Refuse one oversized reusable record rather than letting it monopolize cache storage.
22pub const MAX_ENTRY_BYTES: usize = 16 * 1024 * 1024;
23
24/// A record of the key history, of executions `e`.
25pub(super) const RECORD: &str =
26    "e.completed_at IS NOT NULL AND e.status IN ('GREEN','RED') AND e.key IS NOT NULL";
27
28/// The latest record of a key first: the newest completion, then the most recently stored.
29pub(super) const LATEST: &str = "e.completed_at DESC, e.rowid DESC";
30
31/// One record of a key's history, or with `cache list` the latest record of each key.
32#[derive(Debug, Serialize)]
33#[serde(rename_all = "camelCase")]
34pub struct Entry {
35    pub key: crate::types::ReuseKey,
36    pub eval_def_hash: crate::types::DefinitionHash,
37    pub execution_id: crate::types::ExecutionId,
38    pub verdict: crate::types::ExecutionStatus,
39    #[serde(with = "crate::platform::path_serde")]
40    pub repo_path: std::path::PathBuf,
41    pub eval_id: crate::types::EvalId,
42    /// The target Artifact's fingerprint.
43    pub fingerprint: Option<crate::types::Fingerprint>,
44    pub completed_at: Option<crate::types::Timestamp>,
45    /// `user@host` of the machine that produced the result.
46    pub producer: Option<crate::types::ProducerId>,
47    /// The selected profile variant, when the result came from one.
48    pub variant: Option<crate::config::ProfileVariantName>,
49    /// The remote store URL of a mirrored record.
50    pub origin: Option<crate::types::StoreUrl>,
51    /// How many records the key holds.
52    pub records: i64,
53    pub bytes: i64,
54    pub last_used: crate::types::Timestamp,
55}
56
57async fn open(state: &Path, writable: bool) -> Result<Option<NotifyingConnection>, String> {
58    let state = crate::workspace::canonical_target(state).map_err(|e| e.to_string())?;
59    super::receipts::check_files(&state)?;
60    let path = state.join(DATABASE);
61    if !path.try_exists().map_err(|e| e.to_string())? {
62        return Ok(None);
63    }
64    let connection = Connection::open_with_flags(
65        path,
66        if writable {
67            OpenFlags::SQLITE_OPEN_READ_WRITE
68        } else {
69            OpenFlags::SQLITE_OPEN_READ_ONLY
70        },
71    )
72    .await
73    .map_err(|e| e.to_string())?;
74    let connection = NotifyingConnection::new(connection, &state);
75    let initialized = connection
76        .call(|db| -> Result<bool, Error> {
77            super::wait_when_busy(db)?;
78            let transaction = db.transaction()?;
79            schema_initialized(&transaction)
80        })
81        .await
82        .map_err(|e| e.to_string());
83    match initialized {
84        Ok(true) => Ok(Some(connection)),
85        result => connection.finish(result.map(|_| None)).await,
86    }
87}
88
89/// The latest record of every key, or with `history` every record, latest first per key.
90pub async fn list(state: &Path, history: bool) -> Result<Vec<Entry>, String> {
91    let Some(connection) = open(state, false).await? else {
92        return Ok(Vec::new());
93    };
94    let result = connection
95        .call(move |db| -> Result<_, Error> {
96            let mut statement = db.prepare(&format!(
97                "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"
98            ))?;
99            Ok(statement
100                .query_map([history], |row| {
101                    let id: crate::types::ExecutionId = row.get(2)?;
102                    let data: String = row.get(13)?;
103                    let Some(execution) =
104                        super::unreadable::evidence::<Execution>("execution", &data)
105                    else {
106                        return Ok(None);
107                    };
108                    Ok(Some(Entry {
109                        key: row.get(0)?,
110                        eval_def_hash: row.get(1)?,
111                        execution_id: id,
112                        verdict: row.get(3)?,
113                        repo_path: row.get::<_, String>(4)?.into(),
114                        eval_id: row.get(5)?,
115                        fingerprint: row.get(6)?,
116                        completed_at: row.get(7)?,
117                        producer: execution.producer.map(|producer| producer.name),
118                        variant: execution.options.variant,
119                        origin: row.get(8)?,
120                        records: row.get(9)?,
121                        bytes: row.get(10)?,
122                        last_used: row.get(11)?,
123                    }))
124                })?
125                .filter_map(|row| row.transpose())
126                .collect::<Result<_, _>>()?)
127        })
128        .await
129        .map_err(|e| e.to_string());
130    connection.finish(result).await
131}
132
133/// A key's records, latest first; without `history` only the latest.
134pub async fn show(state: &Path, key: &str, history: bool) -> Result<Vec<Execution>, String> {
135    let Some(connection) = open(state, false).await? else {
136        return Ok(Vec::new());
137    };
138    let Ok(key) = key.parse::<crate::types::ReuseKey>() else {
139        return connection.finish(Ok(Vec::new())).await;
140    };
141    let result = connection
142        .call(move |db| -> Result<_, Error> {
143            let mut statement = db.prepare(&format!(
144                "SELECT e.data FROM executions e WHERE e.key=? AND {RECORD} ORDER BY {LATEST} LIMIT ?"
145            ))?;
146            statement
147                .query_map(params![key, if history { -1 } else { 1 }], |row| {
148                    row.get::<_, String>(0)
149                })?
150                .filter_map(|row| match row {
151                    Ok(data) => super::unreadable::evidence("execution", &data).map(Ok),
152                    Err(error) => Some(Err(Error::Sql(error))),
153                })
154                .collect()
155        })
156        .await
157        .map_err(|e| e.to_string());
158    connection.finish(result).await
159}
160
161/// The latest locally produced record (never a mirror) of each key after `after`, in key order.
162pub async fn local(
163    state: &Path,
164    after: Option<crate::types::ReuseKey>,
165    limit: usize,
166) -> Result<Vec<Execution>, String> {
167    let Some(connection) = open(state, false).await? else {
168        return Ok(Vec::new());
169    };
170    let result = connection
171        .call(move |db| -> Result<_, Error> {
172            let mut statement = db.prepare(&format!(
173                "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 ?"
174            ))?;
175            statement
176                .query_map(
177                    params![
178                        after.as_ref().map_or("", crate::types::ReuseKey::as_str),
179                        limit as i64
180                    ],
181                    |row| row.get::<_, String>(0),
182                )?
183                .filter_map(|row| match row {
184                    Ok(data) => super::unreadable::evidence("execution", &data).map(Ok),
185                    Err(error) => Some(Err(Error::Sql(error))),
186                })
187                .collect()
188        })
189        .await
190        .map_err(|e| e.to_string());
191    connection.finish(result).await
192}
193
194/// Remove every record of an unused key from its history, preserving executions and Runs.
195pub async fn remove(state: &Path, key: &str) -> Result<bool, String> {
196    let Some(connection) = open(state, true).await? else {
197        return Ok(false);
198    };
199    let Ok(key) = key.parse::<crate::types::ReuseKey>() else {
200        return connection.finish(Ok(false)).await;
201    };
202    let result = connection
203        .call(move |db| -> Result<bool, Error> {
204            let transaction =
205                db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
206            let in_use: bool = transaction.query_row(
207                "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'))",
208                [&key],
209                |row| row.get(0),
210            )?;
211            if in_use {
212                return Err(Error::Invalid(format!(
213                    "Key {key} is in use by an active execution or waiter."
214                )));
215            }
216            let removed = transaction.execute(
217                "UPDATE executions SET completed_at=NULL,bytes=NULL,last_used=NULL WHERE key=? AND completed_at IS NOT NULL",
218                [&key],
219            )?;
220            transaction.commit()?;
221            Ok(removed != 0)
222        })
223        .await
224        .map_err(|e| e.to_string());
225    connection.finish(result).await
226}
227
228/// The history columns of a completed record: its sortable completion time and size, or
229/// `None` when it does not join its key's history (no key, no verdict, or over the entry
230/// limit).
231pub(super) fn columns(
232    execution: &Execution,
233    bytes: usize,
234) -> Result<Option<(crate::types::Timestamp, i64)>, Error> {
235    let (Some(_), Some(_), Some(completed_at)) = (
236        &execution.key,
237        execution.verdict(),
238        &execution.completed_at(),
239    ) else {
240        return Ok(None);
241    };
242    if bytes > MAX_ENTRY_BYTES {
243        return Ok(None);
244    }
245    Ok(Some((*completed_at, bytes as i64)))
246}
247
248/// Mark a reused record as used now.
249pub(super) fn touch(
250    db: &rusqlite::Connection,
251    execution_id: &crate::types::ExecutionId,
252    at: crate::types::Timestamp,
253) -> Result<(), Error> {
254    db.execute(
255        "UPDATE executions SET last_used=? WHERE id=? AND completed_at IS NOT NULL",
256        params![at, execution_id],
257    )?;
258    Ok(())
259}
260
261/// The caps [`collect`] keeps the history no checkout reuses now within.
262#[derive(Debug, Clone, Copy)]
263struct Caps {
264    entries: i64,
265    bytes: i64,
266}
267
268const CAPS: Caps = Caps {
269    entries: MAX_ENTRIES,
270    bytes: MAX_BYTES,
271};
272
273/// Collect the key history once a Run has finished. The newest record of each eval in each
274/// checkout that still exists is what that checkout's next verify reuses, so it stays whatever
275/// the caps: one large project evicts neither another project's results nor its own. The
276/// other records stay within [`MAX_ENTRIES`] and [`MAX_BYTES`], the least recently used
277/// leaving first and Human sign-offs last. Records of a key with an active execution, and
278/// records an in-flight request waits for, stay as well.
279pub(crate) async fn collect(state: &Path) -> Result<(), String> {
280    let Some(connection) = open(state, true).await? else {
281        return Ok(());
282    };
283    let result = connection
284        .call(|db| collect_within(db, CAPS))
285        .await
286        .map_err(|e| e.to_string());
287    connection.finish(result).await
288}
289
290/// A record as collection weighs it.
291struct Record {
292    id: crate::types::ExecutionId,
293    checkout: Option<String>,
294    eval: Option<String>,
295    human: bool,
296    bytes: i64,
297    last_used: String,
298    held: bool,
299}
300
301fn collect_within(db: &mut rusqlite::Connection, caps: Caps) -> Result<(), Error> {
302    let transaction = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
303    // Latest first: the first record seen of an eval in a checkout is the one it reuses.
304    let records = transaction
305        .prepare(&format!(
306            "SELECT e.id,json_extract(e.data,'$.provenance.repoPath'),json_extract(e.data,'$.provenance.evalId'),
307             coalesce(json_extract(e.data,'$.profile.kind')='human',0),e.bytes,e.last_used,
308             EXISTS(SELECT 1 FROM executions a WHERE a.key=e.key AND a.status IN ('RUNNING','WAITING_HUMAN'))
309             OR EXISTS(SELECT 1 FROM requests q WHERE q.execution_id=e.id AND q.status IN ('QUEUED','RUNNING','WAITING_HUMAN'))
310             FROM executions e WHERE {RECORD} ORDER BY {LATEST}"
311        ))?
312        .query_map([], |row| {
313            Ok(Record {
314                id: row.get(0)?,
315                checkout: row.get(1)?,
316                eval: row.get(2)?,
317                human: row.get(3)?,
318                bytes: row.get(4)?,
319                last_used: row.get(5)?,
320                held: row.get(6)?,
321            })
322        })?
323        .collect::<Result<Vec<_>, _>>()?;
324    let mut reused = std::collections::BTreeSet::new();
325    let mut present = std::collections::BTreeMap::new();
326    let mut history = Vec::new();
327    let mut evicted = Vec::new();
328    for record in records {
329        let newest = match (&record.checkout, &record.eval) {
330            (Some(checkout), Some(eval)) => {
331                reused.insert((checkout.clone(), eval.clone()))
332                    && *present
333                        .entry(checkout.clone())
334                        .or_insert_with(|| Path::new(checkout).is_dir())
335            }
336            _ => false,
337        };
338        if record.held {
339            continue;
340        }
341        if record.bytes > MAX_ENTRY_BYTES as i64 {
342            evicted.push(record.id);
343        } else if !newest {
344            history.push(record);
345        }
346    }
347    // Human sign-offs leave last; among the rest the least recently used leave first.
348    history.sort_by(|a, b| {
349        b.human
350            .cmp(&a.human)
351            .then_with(|| b.last_used.cmp(&a.last_used))
352    });
353    let (mut entries, mut bytes) = (0, 0);
354    for record in history {
355        if entries < caps.entries && bytes + record.bytes <= caps.bytes {
356            entries += 1;
357            bytes += record.bytes;
358        } else {
359            evicted.push(record.id);
360        }
361    }
362    {
363        // Execution and receipt rows stay; only their place in the history goes.
364        let mut evict = transaction.prepare(
365            "UPDATE executions SET completed_at=NULL,bytes=NULL,last_used=NULL WHERE id=?",
366        )?;
367        for id in &evicted {
368            evict.execute([id])?;
369        }
370    }
371    Ok(transaction.commit()?)
372}
373
374#[cfg(test)]
375mod tests;
artifactize_core/store/history.rs

artifactize_core/store/
history.rs
