artifactize_core/
store.rs

1//! The state database and storage maintenance.
2
3use std::path::{Path, PathBuf};
4
5use crate::{platform, workspace::canonical_target};
6
7mod blocked;
8pub mod definitions;
9mod executions;
10pub(crate) mod history;
11mod human;
12pub use human::HumanClaim;
13mod catalog;
14pub(crate) mod connection;
15pub mod prune;
16pub(crate) mod receipts;
17mod result;
18mod run_record;
19mod runs;
20mod snapshots;
21mod touched;
22mod unreadable;
23mod validation;
24pub use unreadable::{Unreadable, read_unreadable};
25mod lifecycle;
26pub(crate) mod migrations;
27pub(crate) mod run_control;
28mod run_read;
29pub use lifecycle::{ExecutionState, GreenResult, PendingRequest, RequestState, RunState};
30pub use result::{ExecutionResult, Field as ResultField, RuntimeResult};
31pub use run_control::read_run_repo;
32pub use run_read::read_run;
33pub(crate) use run_read::{read_new_results, read_run_record, read_run_requests_only};
34pub use snapshots::{
35    ArtifactValidation, Blocker, ChildIdentity, EvalValidation, HumanDefinition, Validation,
36    ValidationSnapshot,
37};
38mod wait_timeout;
39pub use catalog::{CatalogRun, Signoff, read_catalog};
40pub use runs::{RunSummary, read_runs, read_scoped_runs};
41pub(crate) mod reader;
42mod requests;
43pub(crate) use executions::read_profiled_executions;
44pub use executions::{
45    Capacity, Claim, Execution, ExecutionOptions, Origin, Producer, Provenance,
46    read_keyed_executions, read_latest_cached,
47};
48#[cfg(any(test, feature = "test-support"))]
49pub use receipts::OwnedReceipts;
50pub(crate) use receipts::regular_files as check_probe_files;
51pub use receipts::{
52    DATABASE, EARLIER_STATE, LastRequest, QueueReason, Receipts, Request, Run, RunView,
53    StoppedBackend, read_latest_requests, read_state_id, schema_error, state_schema,
54};
55pub use requests::read_request_repo;
56pub use requests::{
57    RequestView, Shown, read_original, read_original_as, read_request, read_request_as,
58    read_request_batch, read_requests, read_requests_in, read_run_requests, read_session_request,
59    read_waiting, read_waiting_originals,
60};
61
62/// How long one database operation waits for the state other processes are writing before it
63/// fails. Several Runs, the monitor and MCP servers share one state; a Run's write waits behind
64/// theirs rather than failing the Run, while a lock that never frees still ends in an error.
65pub(crate) const SQLITE_BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
66
67/// How often a busy operation tries again. SQLite's own busy timeout backs off to 100 ms
68/// between tries, so under steady writes from other connections, which try every 1–2 ms when
69/// they start waiting, a long waiter rarely finds the lock free and runs out its time while
70/// the others commit (#372). Trying at this pace keeps every waiter in the race.
71const SQLITE_BUSY_POLL: std::time::Duration = std::time::Duration::from_millis(1);
72
73/// Make `db` wait for a busy database, trying every [`SQLITE_BUSY_POLL`] for up to
74/// [`SQLITE_BUSY_TIMEOUT`].
75pub(crate) fn wait_when_busy(db: &rusqlite::Connection) -> rusqlite::Result<()> {
76    db.busy_handler(Some(busy))
77}
78
79/// SQLite's busy handler: `tries` counts the earlier calls for this one wait, which runs on
80/// the calling thread, so the wait's start is kept per thread.
81fn busy(tries: i32) -> bool {
82    thread_local! {
83        static SINCE: std::cell::Cell<std::time::Instant> = std::cell::Cell::new(std::time::Instant::now());
84    }
85    let now = std::time::Instant::now();
86    if tries == 0 {
87        SINCE.set(now);
88    }
89    if now.duration_since(SINCE.get()) >= SQLITE_BUSY_TIMEOUT {
90        return false;
91    }
92    std::thread::sleep(SQLITE_BUSY_POLL);
93    true
94}
95
96/// Allow four short backoffs for a pending-delete handle to close, bounding the extra
97/// latency of a real file-access failure to 80ms. This is a retry policy, not a test deadline.
98const TRANSIENT_ACCESS_RETRIES: u32 = 5;
99const TRANSIENT_ACCESS_BACKOFF: std::time::Duration = std::time::Duration::from_millis(20);
100
101/// Retry `attempt` while `transient` says its error can clear on its own, before giving up
102/// with its last error. `transient` decides alone: a non-transient error, or the bound run out,
103/// both stop the loop and return that error immediately, never retried further.
104fn retry<T, E>(
105    mut attempt: impl FnMut() -> Result<T, E>,
106    transient: impl Fn(&E) -> bool,
107    mut before_retry: impl FnMut(),
108) -> Result<T, E> {
109    for remaining in (0..TRANSIENT_ACCESS_RETRIES).rev() {
110        match attempt() {
111            Ok(value) => return Ok(value),
112            Err(error) if remaining > 0 && transient(&error) => {
113                before_retry();
114            }
115            Err(error) => return Err(error),
116        }
117    }
118    unreachable!("the loop above always returns by its last attempt")
119}
120
121/// Retry `attempt` past [`platform::transient_file_access`] (Windows transiently denying a
122/// metadata or open probe while another handle finishes closing a just-deleted WAL sidecar)
123/// before giving up with its last error. Every read-only state open reaches this through
124/// [`receipts::check_files`]/[`check_probe_files`], which every reader calls first, so this is
125/// the one place the retry needs to live.
126pub(crate) fn retry_transient_access<T>(
127    attempt: impl FnMut() -> std::io::Result<T>,
128    before_retry: impl FnMut(),
129) -> std::io::Result<T> {
130    retry(attempt, platform::transient_file_access, before_retry)
131}
132
133#[cfg(test)]
134mod retry_tests {
135    use std::{cell::Cell, io};
136
137    #[test]
138    fn file_probe_returns_success_and_permanent_errors_without_backoff() {
139        let calls = Cell::new(0);
140        let result = super::retry_transient_access(
141            || {
142                calls.set(calls.get() + 1);
143                Ok(7)
144            },
145            || panic!("successful probes need no backoff"),
146        );
147        assert_eq!(result.unwrap(), 7);
148        assert_eq!(calls.get(), 1);
149
150        let calls = Cell::new(0);
151        let result = super::retry_transient_access(
152            || {
153                calls.set(calls.get() + 1);
154                Err::<(), _>(io::Error::other("permanent probe failure"))
155            },
156            || panic!("permanent errors cannot be retried"),
157        );
158        assert_eq!(result.unwrap_err().to_string(), "permanent probe failure");
159        assert_eq!(calls.get(), 1);
160    }
161}
162
163/// Resolve the single state directory without creating it.
164pub fn state_dir(explicit: Option<&Path>) -> Result<PathBuf, String> {
165    let path = match explicit {
166        Some(path) => path.to_path_buf(),
167        None => state_home().map_err(|e| e.to_string())?,
168    };
169    canonical_target(&path).map_err(|e| e.to_string())
170}
171
172/// SQLite `user_version`; schema 9 keeps a Run's definitions in their own row, written once,
173/// and numbers every change of a Run, request or execution row; schema 10 adds the open
174/// WAITING_HUMAN Run and names a Run that ended with only sign-offs left YELLOW. Versions from
175/// schema 6 upgrade additively.
176pub const STATE_SCHEMA_VERSION: u32 = 10;
177
178#[derive(Debug, thiserror::Error)]
179#[error("cannot resolve state home: set {}", state_home_variables())]
180pub struct StateHomeError;
181
182/// The variables `state_home` reads, in order, for the error that names them.
183fn state_home_variables() -> String {
184    let names: Vec<&str> = std::iter::once("ARTIFACTIZE_STATE_HOME")
185        .chain(platform::STATE_VARIABLES.iter().copied())
186        .chain([platform::HOME_VARIABLE])
187        .collect();
188    let (last, rest) = names.split_last().expect("constant names");
189    format!("{}, or {last}", rest.join(", "))
190}
191
192/// Resolve the default state home without creating directories; empty variables are ignored.
193/// The platform's state variables come after `ARTIFACTIZE_STATE_HOME` and before the home.
194pub fn state_home() -> Result<PathBuf, StateHomeError> {
195    platform::state_directory().ok_or(StateHomeError)
196}
197
198#[cfg(test)]
199#[path = "store/file_tests.rs"]
200mod file_tests;
artifactize_core/store.rs

artifactize_core/
store.rs
