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
7pub mod definitions;
8mod executions;
9pub(crate) mod history;
10mod human;
11pub use human::HumanClaim;
12mod catalog;
13pub mod prune;
14pub(crate) mod receipts;
15mod result;
16mod runs;
17mod snapshots;
18mod unreadable;
19mod validation;
20pub use unreadable::{Unreadable, read_unreadable};
21mod lifecycle;
22pub use lifecycle::{ExecutionState, PendingRequest, RequestState, RunState};
23pub use result::{ExecutionResult, Field as ResultField, RuntimeResult};
24pub use snapshots::{ArtifactValidation, Blocker, ChildIdentity, HumanDefinition, Validation};
25mod wait_timeout;
26pub use catalog::{CatalogRun, Signoff, read_catalog};
27pub use runs::{RunSummary, read_runs, read_scoped_runs};
28pub(crate) mod reader;
29mod requests;
30pub use executions::{
31    Capacity, Claim, Execution, ExecutionOptions, Origin, Producer, Provenance,
32    read_keyed_executions, read_latest_cached,
33};
34pub(crate) use receipts::regular_files as check_probe_files;
35pub use receipts::{
36    DATABASE, EARLIER_STATE, LastRequest, QueueReason, Receipts, Request, Run, RunView,
37    StoppedBackend, read_latest_requests, read_run, read_state_id, schema_error, state_schema,
38};
39pub(crate) use requests::read_request_repo;
40pub use requests::{
41    RequestView, read_original, read_request, read_request_batch, read_requests, read_run_requests,
42    read_session_request, read_waiting, read_waiting_originals,
43};
44
45/// Let concurrent readers/writers finish short SQLite transactions without an
46/// immediate busy error, while limiting how long one database operation can block.
47pub(crate) const SQLITE_BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
48
49/// Allow four short backoffs for a pending-delete handle to close, bounding the extra
50/// latency of a real file-access failure to 80ms. This is a retry policy, not a test deadline.
51const TRANSIENT_ACCESS_RETRIES: u32 = 5;
52const TRANSIENT_ACCESS_BACKOFF: std::time::Duration = std::time::Duration::from_millis(20);
53
54/// Retry `attempt` while `transient` says its error can clear on its own, before giving up
55/// with its last error. `transient` decides alone: a non-transient error, or the bound run out,
56/// both stop the loop and return that error immediately, never retried further.
57fn retry<T, E>(
58    mut attempt: impl FnMut() -> Result<T, E>,
59    transient: impl Fn(&E) -> bool,
60    mut before_retry: impl FnMut(),
61) -> Result<T, E> {
62    for remaining in (0..TRANSIENT_ACCESS_RETRIES).rev() {
63        match attempt() {
64            Ok(value) => return Ok(value),
65            Err(error) if remaining > 0 && transient(&error) => {
66                before_retry();
67            }
68            Err(error) => return Err(error),
69        }
70    }
71    unreachable!("the loop above always returns by its last attempt")
72}
73
74/// Retry `attempt` past [`platform::transient_file_access`] (Windows transiently denying a
75/// metadata or open probe while another handle finishes closing a just-deleted WAL sidecar)
76/// before giving up with its last error. Every read-only state open reaches this through
77/// [`receipts::check_files`]/[`check_probe_files`], which every reader calls first, so this is
78/// the one place the retry needs to live.
79pub(crate) fn retry_transient_access<T>(
80    attempt: impl FnMut() -> std::io::Result<T>,
81    before_retry: impl FnMut(),
82) -> std::io::Result<T> {
83    retry(attempt, platform::transient_file_access, before_retry)
84}
85
86#[cfg(test)]
87mod retry_tests {
88    use std::cell::Cell;
89
90    #[test]
91    fn retry_succeeds_within_bound_gives_up_past_it_and_never_retries_a_non_transient_error() {
92        let calls = Cell::new(0);
93        let waits = Cell::new(0);
94        let result = super::retry(
95            || {
96                calls.set(calls.get() + 1);
97                if calls.get() <= 2 {
98                    Err::<u32, _>("transient")
99                } else {
100                    Ok(calls.get())
101                }
102            },
103            |error: &&str| *error == "transient",
104            || waits.set(waits.get() + 1),
105        );
106        assert_eq!(result, Ok(3));
107        assert_eq!(calls.get(), 3);
108        assert_eq!(waits.get(), 2);
109
110        let calls = Cell::new(0);
111        let waits = Cell::new(0);
112        let result = super::retry(
113            || {
114                calls.set(calls.get() + 1);
115                Err::<(), _>("transient")
116            },
117            |error: &&str| *error == "transient",
118            || waits.set(waits.get() + 1),
119        );
120        assert_eq!(result, Err("transient"));
121        assert_eq!(calls.get(), super::TRANSIENT_ACCESS_RETRIES);
122        assert_eq!(waits.get(), 4);
123
124        let calls = Cell::new(0);
125        let waits = Cell::new(0);
126        let result = super::retry(
127            || {
128                calls.set(calls.get() + 1);
129                Err::<(), _>("permanent")
130            },
131            |error: &&str| *error == "transient",
132            || waits.set(waits.get() + 1),
133        );
134        assert_eq!(result, Err("permanent"));
135        assert_eq!(calls.get(), 1);
136        assert_eq!(waits.get(), 0);
137    }
138}
139
140/// Resolve the single state directory without creating it.
141pub fn state_dir(explicit: Option<&Path>) -> Result<PathBuf, String> {
142    let path = match explicit {
143        Some(path) => path.to_path_buf(),
144        None => state_home().map_err(|e| e.to_string())?,
145    };
146    canonical_target(&path).map_err(|e| e.to_string())
147}
148
149/// SQLite `user_version`; dependency requests and derived verdicts require schema 6.
150/// Earlier state is not migrated.
151pub const STATE_SCHEMA_VERSION: u32 = 6;
152
153#[derive(Debug, thiserror::Error)]
154#[error("cannot resolve state home: set {}", state_home_variables())]
155pub struct StateHomeError;
156
157/// The variables `state_home` reads, in order, for the error that names them.
158fn state_home_variables() -> String {
159    let names: Vec<&str> = std::iter::once("ARTIFACTIZE_STATE_HOME")
160        .chain(platform::STATE_VARIABLES.iter().copied())
161        .chain([platform::HOME_VARIABLE])
162        .collect();
163    let (last, rest) = names.split_last().expect("constant names");
164    format!("{}, or {last}", rest.join(", "))
165}
166
167/// Resolve the default state home without creating directories; empty variables are ignored.
168/// The platform's state variables come after `ARTIFACTIZE_STATE_HOME` and before the home.
169pub fn state_home() -> Result<PathBuf, StateHomeError> {
170    platform::state_directory().ok_or(StateHomeError)
171}
172
173#[cfg(test)]
174#[path = "store/file_tests.rs"]
175mod file_tests;
artifactize_core/store.rs

artifactize_core/
store.rs
