artifactize_core/store/
receipts.rs

1use crate::types::{
2    Fingerprint, RequestId, RequestStatus, ReuseKey, RunId, RunStatus, SessionId, Timestamp,
3};
4use std::{
5    collections::{BTreeMap, BTreeSet},
6    path::{Path, PathBuf},
7    time::Duration,
8};
9
10use rusqlite::{OptionalExtension, params};
11use serde::{Deserialize, Serialize};
12
13use tokio_rusqlite::Connection;
14
15use super::{ExecutionOptions, Origin, Producer, Provenance, STATE_SCHEMA_VERSION};
16use crate::{
17    config::ProfileKind,
18    workspace::{canonical_target, outside_workspace, prepare_directory},
19};
20
21pub const DATABASE: &str = "state.sqlite";
22/// Yield between WAL initialization races instead of spinning, still within the
23/// shared SQLite contention deadline used by the outer initialization loop.
24const INITIALIZATION_RETRY_INTERVAL: Duration = Duration::from_millis(20);
25
26/// The four tables of a state and the indexes its lookups use.
27///
28/// - `runs`: one row per Run.
29/// - `requests`: one row per eval of a Run, at its `ordinal` in the Run's selection order;
30///   `claimed_by` and `claimed_at` hold the Human claim of a waiting request.
31/// - `executions`: one row per execution. `backend` names the Agent backend it runs on, and
32///   a RUNNING execution with a backend holds one of that backend's machine-wide slots. A
33///   completed GREEN/RED execution with a key is a record of the key's history while
34///   `completed_at` (sortable), `bytes` and `last_used` are set; the cache GC and `cache rm`
35///   clear them, and the execution stays.
36/// - `state_meta`: named values of the state itself: `id`, a random UUID made when the
37///   database is created, which identifies the state in Agent session references.
38const SCHEMA: &str =
39    "CREATE TABLE runs(id TEXT PRIMARY KEY, repo TEXT NOT NULL, status TEXT NOT NULL, data TEXT NOT NULL, request_key TEXT);
40    CREATE UNIQUE INDEX run_request_key ON runs(repo,request_key) WHERE request_key IS NOT NULL;
41    CREATE TABLE requests(id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES runs(id), eval_id TEXT NOT NULL, ordinal INTEGER NOT NULL, execution_id TEXT REFERENCES executions(id), status TEXT NOT NULL, claimed_by TEXT, claimed_at TEXT, data TEXT NOT NULL, UNIQUE(run_id, eval_id), UNIQUE(run_id, ordinal));
42    CREATE TABLE executions(id TEXT PRIMARY KEY, key TEXT, eval_def_hash TEXT NOT NULL, status TEXT NOT NULL, owner_pid INTEGER NOT NULL, owner_start_time INTEGER NOT NULL, backend TEXT, profile_identity TEXT, completed_at TEXT, bytes INTEGER, last_used TEXT, data TEXT NOT NULL);
43    CREATE TABLE state_meta(name TEXT PRIMARY KEY, value TEXT NOT NULL);
44    CREATE INDEX active_request_execution ON requests(execution_id) WHERE status IN ('QUEUED','RUNNING','WAITING_HUMAN');
45    CREATE UNIQUE INDEX active_key ON executions(key,profile_identity) WHERE key IS NOT NULL AND status IN ('RUNNING','WAITING_HUMAN');
46    CREATE INDEX key_history ON executions(key, completed_at);
47    CREATE INDEX definition_history ON executions(eval_def_hash, completed_at);
48    CREATE INDEX running_backend ON executions(backend) WHERE status='RUNNING';";
49
50/// Versions before the supported additive upgrade boundary are refused.
51pub const EARLIER_STATE: &str = "This state was written by an earlier artifactize. Start a new state (set ARTIFACTIZE_STATE_HOME or move the old one away). Only schema 6 and later supported versions can be upgraded.";
52
53/// Why a state database of schema `version`, neither this one nor uninitialized, is refused.
54pub fn schema_error(version: u32) -> String {
55    match version {
56        0 => "Unsupported state schema version: 0.".into(),
57        version if version < STATE_SCHEMA_VERSION => EARLIER_STATE.into(),
58        version => {
59            format!("Unsupported state schema version: {version}; a newer artifactize wrote it.")
60        }
61    }
62}
63
64#[derive(Debug, thiserror::Error)]
65pub enum Error {
66    #[error(transparent)]
67    Sql(#[from] rusqlite::Error),
68    #[error(transparent)]
69    Json(#[from] serde_json::Error),
70    #[error("{0}")]
71    Invalid(String),
72}
73
74#[derive(Debug, Clone, Serialize, Deserialize)]
75#[serde(try_from = "RunWire", into = "RunWire")]
76pub struct Run {
77    pub id: RunId,
78    pub repo_path: PathBuf,
79    /// Optional display metadata; schema 5 Runs written before this remain readable.
80    pub repository: crate::repository::Identity,
81    pub state_dir: PathBuf,
82    pub state: super::RunState,
83    pub modes: crate::runs::RunModes,
84    pub control: Option<crate::runs::Control>,
85    pub request_key: Option<crate::types::RequestKey>,
86    pub created_at: Timestamp,
87    pub selection: crate::project::selection::Selection,
88    pub profile: Option<crate::project::selection::ProfileSelection>,
89    pub definitions: super::definitions::Definitions,
90    pub recursive: bool,
91    pub force: bool,
92    pub jobs: Option<usize>,
93    /// Fingerprints computed at once; absent in Runs saved before 0.5.
94    pub fingerprint_jobs: Option<usize>,
95    pub max_executions: Option<u64>,
96    pub executions_started: u64,
97    pub wait_timeout_ms: Option<Duration>,
98    pub wait_timed_out: bool,
99    /// Kinds whose evals only reuse a result; one with nothing to reuse is not executed.
100    pub reuse_only: BTreeSet<ProfileKind>,
101    pub validation: super::Validation,
102    /// Agent backends this Run stopped admitting reviews on, in the order they stopped.
103    pub stopped_backends: Vec<StoppedBackend>,
104    /// Verdicts the Run took from saved results for evals it has no request for, such as the
105    /// dependencies of a partial Run, so readers judge gates on the Run's own evidence.
106    pub evidence: BTreeMap<crate::types::EvalId, crate::types::RequestStatus>,
107}
108
109#[derive(Debug, Clone, Serialize, Deserialize)]
110#[serde(rename_all = "camelCase")]
111pub(crate) struct RunWire {
112    pub id: RunId,
113    #[serde(with = "crate::platform::path_serde")]
114    pub repo_path: PathBuf,
115    /// Optional display metadata; schema 5 Runs written before this remain readable.
116    #[serde(flatten)]
117    pub repository: crate::repository::Identity,
118    #[serde(with = "crate::platform::path_serde")]
119    pub state_dir: PathBuf,
120    #[serde(default, skip_serializing_if = "crate::runs::RunModes::is_default")]
121    pub modes: crate::runs::RunModes,
122    #[serde(default, skip_serializing_if = "Option::is_none")]
123    pub control: Option<crate::runs::Control>,
124    #[serde(default, skip_serializing_if = "Option::is_none")]
125    pub request_key: Option<crate::types::RequestKey>,
126    pub status: RunStatus,
127    pub created_at: Timestamp,
128    pub completed_at: Option<Timestamp>,
129    pub selection: crate::project::selection::Selection,
130    #[serde(default)]
131    pub profile: Option<crate::project::selection::ProfileSelection>,
132    #[serde(default)]
133    pub definitions: super::definitions::Definitions,
134    #[serde(default)]
135    pub recursive: bool,
136    #[serde(default)]
137    pub force: bool,
138    #[serde(default, skip_serializing_if = "Option::is_none")]
139    pub jobs: Option<usize>,
140    /// Fingerprints computed at once; absent in Runs saved before 0.5.
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub fingerprint_jobs: Option<usize>,
143    #[serde(default)]
144    pub max_executions: Option<u64>,
145    #[serde(default)]
146    pub executions_started: u64,
147    #[serde(default, with = "super::wait_timeout")]
148    pub wait_timeout_ms: Option<Duration>,
149    #[serde(default)]
150    pub wait_timed_out: bool,
151    /// Kinds whose evals only reuse a result; one with nothing to reuse is not executed.
152    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
153    pub reuse_only: BTreeSet<ProfileKind>,
154    pub validation: super::Validation,
155    pub error: Option<String>,
156    /// Agent backends this Run stopped admitting reviews on, in the order they stopped.
157    #[serde(default, skip_serializing_if = "Vec::is_empty")]
158    pub stopped_backends: Vec<StoppedBackend>,
159    /// Verdicts the Run took from saved results for evals it has no request for, such as the
160    /// dependencies of a partial Run, so readers judge gates on the Run's own evidence.
161    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
162    pub evidence: BTreeMap<crate::types::EvalId, crate::types::RequestStatus>,
163}
164
165impl TryFrom<RunWire> for Run {
166    type Error = String;
167    fn try_from(wire: RunWire) -> Result<Self, Self::Error> {
168        let state = super::RunState::from_parts(wire.status, wire.completed_at, wire.error)?;
169        Ok(Self {
170            state,
171            modes: wire.modes,
172            control: wire.control,
173            request_key: wire.request_key,
174            id: wire.id,
175            repo_path: wire.repo_path,
176            repository: wire.repository,
177            state_dir: wire.state_dir,
178            created_at: wire.created_at,
179            selection: wire.selection,
180            profile: wire.profile,
181            definitions: wire.definitions,
182            recursive: wire.recursive,
183            force: wire.force,
184            jobs: wire.jobs,
185            fingerprint_jobs: wire.fingerprint_jobs,
186            max_executions: wire.max_executions,
187            executions_started: wire.executions_started,
188            wait_timeout_ms: wire.wait_timeout_ms,
189            wait_timed_out: wire.wait_timed_out,
190            reuse_only: wire.reuse_only,
191            validation: wire.validation,
192            stopped_backends: wire.stopped_backends,
193            evidence: wire.evidence,
194        })
195    }
196}
197impl From<Run> for RunWire {
198    fn from(record: Run) -> Self {
199        Self {
200            modes: record.modes.clone(),
201            control: record.control.clone(),
202            request_key: record.request_key.clone(),
203            status: record.status(),
204            completed_at: record.completed_at(),
205            error: record.error().map(str::to_owned),
206            id: record.id,
207            repo_path: record.repo_path,
208            repository: record.repository,
209            state_dir: record.state_dir,
210            created_at: record.created_at,
211            selection: record.selection,
212            profile: record.profile,
213            definitions: record.definitions,
214            recursive: record.recursive,
215            force: record.force,
216            jobs: record.jobs,
217            fingerprint_jobs: record.fingerprint_jobs,
218            max_executions: record.max_executions,
219            executions_started: record.executions_started,
220            wait_timeout_ms: record.wait_timeout_ms,
221            wait_timed_out: record.wait_timed_out,
222            reuse_only: record.reuse_only,
223            validation: record.validation,
224            stopped_backends: record.stopped_backends,
225            evidence: record.evidence,
226        }
227    }
228}
229impl Run {
230    pub fn status(&self) -> RunStatus {
231        self.state.status()
232    }
233    pub fn completed_at(&self) -> Option<Timestamp> {
234        self.state.completed_at()
235    }
236    pub fn error(&self) -> Option<&str> {
237        self.state.error()
238    }
239}
240
241/// An Agent backend a Run stopped after an AUTHENTICATION or QUOTA failure, which every
242/// later review on it would repeat.
243#[derive(Debug, Clone, Serialize, Deserialize)]
244#[serde(rename_all = "camelCase")]
245pub struct StoppedBackend {
246    pub backend: crate::config::Backend,
247    pub error_code: crate::types::BackendStopCode,
248    pub eval_id: crate::types::EvalId,
249    pub request_id: RequestId,
250    pub error: String,
251}
252
253#[derive(Debug, Clone, Serialize, Deserialize)]
254#[serde(try_from = "RequestWire", into = "RequestWire")]
255pub struct Request {
256    pub state: super::RequestState,
257    pub id: RequestId,
258    pub run_id: RunId,
259    pub eval_id: crate::types::EvalId,
260    pub target: crate::types::ArtifactName,
261    pub title: String,
262    pub profile: crate::config::StoredProfile,
263    pub requested_profile: crate::config::StoredProfile,
264    pub eval_def_hash: crate::types::DefinitionHash,
265    /// The requested execution options, or a reused result's.
266    pub options: ExecutionOptions,
267    pub execution_id: Option<crate::types::ExecutionId>,
268    pub provenance: Option<Provenance>,
269    /// Usage spent by this request; a reused request spent none.
270    pub usage: Option<Vec<crate::llm::Attempt>>,
271    /// The reused execution's original usage, never counted as spent.
272    pub reused_usage: Option<Vec<crate::llm::Attempt>>,
273    /// The reused result came from the live execution this request waited for, not a
274    /// completed record. Saved only; output reports it as `source.kind` ([`crate::query::source`]).
275    pub joined: bool,
276    /// Who produced a reused result.
277    pub producer: Option<Producer>,
278    /// The Human claimant who submitted a reused result.
279    pub reviewer: Option<crate::types::ReviewerId>,
280    /// The remote store, publisher and publication time of a result reused from a mirror.
281    pub origin: Option<Origin>,
282    /// The Agent review's session id ([`crate::agent::session_id`]), which every request of
283    /// the review sends as its prompt-cache identity. Absent when no Agent review ran, as on
284    /// reuse, and on requests saved before it existed.
285    pub session_id: Option<SessionId>,
286    /// Where the saved conversation behind this request's result lives: its own review's,
287    /// or for a reused result the producing review's. Absent when none was saved.
288    pub session: Option<crate::agent::session::SessionRef>,
289    pub human_definition: Option<super::HumanDefinition>,
290    pub payload: crate::config::StoredPayload,
291    pub references: BTreeMap<crate::config::ReferenceName, crate::types::ArtifactName>,
292    pub deps: Vec<crate::types::ArtifactName>,
293    pub force: bool,
294    /// The target Artifact's fingerprint.
295    pub fingerprint: Option<Fingerprint>,
296    /// The reuse key; absent without a fingerprint on every Artifact the eval depends on.
297    pub key: Option<ReuseKey>,
298    /// Each Artifact the key covers, with its fingerprint.
299    pub fingerprints: std::collections::BTreeMap<crate::types::ArtifactName, Fingerprint>,
300    pub created_at: Timestamp,
301    pub started_at: Option<Timestamp>,
302    pub cwd: PathBuf,
303    pub run_dir: Option<PathBuf>,
304    pub argv: Option<Vec<String>>,
305    pub child: Option<super::ChildIdentity>,
306    pub blocked_reason: Option<String>,
307    /// Unfulfilled Evals behind this request’s dependency gates.
308    pub blocked_by: Vec<super::Blocker>,
309    /// What a queued request waits for besides a job; `blocked_reason` says it in words.
310    pub queue: Option<QueueReason>,
311}
312
313/// What a queued request waits for besides a job slot of its Run.
314#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
315#[serde(tag = "kind", rename_all = "camelCase")]
316pub enum QueueReason {
317    /// A free machine-wide slot of this backend (`limits.json`).
318    Slot { backend: crate::config::Backend },
319}
320
321#[derive(Debug, Clone, Serialize, Deserialize)]
322#[serde(rename_all = "camelCase")]
323pub(crate) struct RequestWire {
324    pub id: RequestId,
325    pub run_id: RunId,
326    pub eval_id: crate::types::EvalId,
327    pub target: crate::types::ArtifactName,
328    pub title: String,
329    pub profile: crate::config::StoredProfile,
330    pub requested_profile: crate::config::StoredProfile,
331    pub eval_def_hash: crate::types::DefinitionHash,
332    /// The requested execution options, or a reused result's.
333    #[serde(default)]
334    pub options: ExecutionOptions,
335    pub execution_id: Option<crate::types::ExecutionId>,
336    pub provenance: Option<Provenance>,
337    /// Usage spent by this request; a reused request spent none.
338    pub usage: Option<Vec<crate::llm::Attempt>>,
339    /// The reused execution's original usage, never counted as spent.
340    #[serde(default, skip_serializing_if = "Option::is_none")]
341    pub reused_usage: Option<Vec<crate::llm::Attempt>>,
342    /// The reused result came from the live execution this request waited for, not a
343    /// completed record. Saved only; output reports it as `source.kind` ([`crate::query::source`]).
344    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
345    pub joined: bool,
346    /// Who produced a reused result.
347    #[serde(default, skip_serializing_if = "Option::is_none")]
348    pub producer: Option<Producer>,
349    /// The Human claimant who submitted a reused result.
350    #[serde(default, skip_serializing_if = "Option::is_none")]
351    pub reviewer: Option<crate::types::ReviewerId>,
352    /// The remote store, publisher and publication time of a result reused from a mirror.
353    #[serde(default, skip_serializing_if = "Option::is_none")]
354    pub origin: Option<Origin>,
355    /// The Agent review's session id ([`crate::agent::session_id`]), which every request of
356    /// the review sends as its prompt-cache identity. Absent when no Agent review ran, as on
357    /// reuse, and on requests saved before it existed.
358    #[serde(default, skip_serializing_if = "Option::is_none")]
359    pub session_id: Option<SessionId>,
360    /// Where the saved conversation behind this request's result lives: its own review's,
361    /// or for a reused result the producing review's. Absent when none was saved.
362    #[serde(default, skip_serializing_if = "Option::is_none")]
363    pub session: Option<crate::agent::session::SessionRef>,
364    #[serde(default, skip_serializing_if = "Option::is_none")]
365    pub human_definition: Option<super::HumanDefinition>,
366    pub payload: crate::config::StoredPayload,
367    pub references: BTreeMap<crate::config::ReferenceName, crate::types::ArtifactName>,
368    pub deps: Vec<crate::types::ArtifactName>,
369    #[serde(default)]
370    pub force: bool,
371    /// The target Artifact's fingerprint.
372    #[serde(default)]
373    pub fingerprint: Option<Fingerprint>,
374    /// The reuse key; absent without a fingerprint on every Artifact the eval depends on.
375    #[serde(default)]
376    pub key: Option<ReuseKey>,
377    /// Each Artifact the key covers, with its fingerprint.
378    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
379    pub fingerprints: std::collections::BTreeMap<crate::types::ArtifactName, Fingerprint>,
380    pub status: RequestStatus,
381    pub created_at: Timestamp,
382    pub started_at: Option<Timestamp>,
383    pub completed_at: Option<Timestamp>,
384    #[serde(with = "crate::platform::path_serde")]
385    pub cwd: PathBuf,
386    #[serde(default, with = "crate::platform::path_serde::option")]
387    pub run_dir: Option<PathBuf>,
388    pub argv: Option<Vec<String>>,
389    pub child: Option<super::ChildIdentity>,
390    pub result: Option<super::ExecutionResult>,
391    pub error: Option<String>,
392    pub error_code: Option<crate::types::FailureCode>,
393    pub blocked_reason: Option<String>,
394    /// Unfulfilled Evals behind this request’s dependency gates.
395    #[serde(default, skip_serializing_if = "Vec::is_empty")]
396    pub blocked_by: Vec<super::Blocker>,
397    /// Absent from requests saved before it existed, which read as waiting for a job.
398    #[serde(default, skip_serializing_if = "Option::is_none")]
399    pub queue: Option<QueueReason>,
400}
401
402impl TryFrom<RequestWire> for Request {
403    type Error = String;
404    fn try_from(wire: RequestWire) -> Result<Self, Self::Error> {
405        let state = super::RequestState::from_parts(
406            wire.status,
407            wire.result,
408            wire.error,
409            wire.error_code,
410            wire.completed_at,
411        )?;
412        Ok(Self {
413            state,
414            id: wire.id,
415            run_id: wire.run_id,
416            eval_id: wire.eval_id,
417            target: wire.target,
418            title: wire.title,
419            profile: wire.profile,
420            requested_profile: wire.requested_profile,
421            eval_def_hash: wire.eval_def_hash,
422            options: wire.options,
423            execution_id: wire.execution_id,
424            provenance: wire.provenance,
425            usage: wire.usage,
426            reused_usage: wire.reused_usage,
427            joined: wire.joined,
428            producer: wire.producer,
429            reviewer: wire.reviewer,
430            origin: wire.origin,
431            session_id: wire.session_id,
432            session: wire.session,
433            human_definition: wire.human_definition,
434            payload: wire.payload,
435            references: wire.references,
436            deps: wire.deps,
437            force: wire.force,
438            fingerprint: wire.fingerprint,
439            key: wire.key,
440            fingerprints: wire.fingerprints,
441            created_at: wire.created_at,
442            started_at: wire.started_at,
443            cwd: wire.cwd,
444            run_dir: wire.run_dir,
445            argv: wire.argv,
446            child: wire.child,
447            blocked_reason: wire.blocked_reason,
448            blocked_by: wire.blocked_by,
449            queue: wire.queue,
450        })
451    }
452}
453impl From<Request> for RequestWire {
454    fn from(record: Request) -> Self {
455        Self {
456            status: record.status(),
457            result: record.result().cloned(),
458            error: record.error().map(str::to_owned),
459            error_code: record.error_code(),
460            completed_at: record.completed_at(),
461            id: record.id,
462            run_id: record.run_id,
463            eval_id: record.eval_id,
464            target: record.target,
465            title: record.title,
466            profile: record.profile,
467            requested_profile: record.requested_profile,
468            eval_def_hash: record.eval_def_hash,
469            options: record.options,
470            execution_id: record.execution_id,
471            provenance: record.provenance,
472            usage: record.usage,
473            reused_usage: record.reused_usage,
474            joined: record.joined,
475            producer: record.producer,
476            reviewer: record.reviewer,
477            origin: record.origin,
478            session_id: record.session_id,
479            session: record.session,
480            human_definition: record.human_definition,
481            payload: record.payload,
482            references: record.references,
483            deps: record.deps,
484            force: record.force,
485            fingerprint: record.fingerprint,
486            key: record.key,
487            fingerprints: record.fingerprints,
488            created_at: record.created_at,
489            started_at: record.started_at,
490            cwd: record.cwd,
491            run_dir: record.run_dir,
492            argv: record.argv,
493            child: record.child,
494            blocked_reason: record.blocked_reason,
495            blocked_by: record.blocked_by,
496            queue: record.queue,
497        }
498    }
499}
500impl Request {
501    pub fn status(&self) -> crate::types::RequestStatus {
502        self.state.status()
503    }
504    pub fn result(&self) -> Option<&super::ExecutionResult> {
505        self.state.result()
506    }
507    pub fn error(&self) -> Option<&str> {
508        self.state.error()
509    }
510    pub fn error_code(&self) -> Option<crate::types::FailureCode> {
511        self.state.error_code()
512    }
513    pub fn completed_at(&self) -> Option<crate::types::Timestamp> {
514        self.state.completed_at()
515    }
516}
517
518#[derive(Debug, Serialize, Deserialize)]
519pub struct RunView {
520    #[serde(flatten)]
521    pub run: Run,
522    pub requests: Vec<Request>,
523    #[serde(default, skip_serializing_if = "Vec::is_empty")]
524    pub unreadable: Vec<super::Unreadable>,
525}
526
527#[derive(Clone)]
528pub struct Receipts {
529    pub(super) connection: NotifyingConnection,
530}
531
532/// A test writer closes its SQLite worker before the caller's temporary state is removed.
533#[cfg(any(test, feature = "test-support"))]
534pub struct OwnedReceipts(Receipts);
535
536#[cfg(any(test, feature = "test-support"))]
537impl OwnedReceipts {
538    pub async fn open(state: &Path, repo: &Path) -> Result<Self, String> {
539        Receipts::open(state, repo).await.map(Self)
540    }
541}
542
543#[cfg(any(test, feature = "test-support"))]
544impl std::ops::Deref for OwnedReceipts {
545    type Target = Receipts;
546    fn deref(&self) -> &Self::Target {
547        &self.0
548    }
549}
550
551#[cfg(any(test, feature = "test-support"))]
552impl Drop for OwnedReceipts {
553    fn drop(&mut self) {
554        let connection = self.0.connection.inner.clone();
555        // Drop can run on a current-thread Tokio test runtime, including during unwinding.
556        // A separate joined thread avoids nesting runtimes or detaching SQLite teardown.
557        let closed = std::thread::spawn(move || {
558            tokio::runtime::Builder::new_current_thread()
559                .build()
560                .expect("create the test writer's teardown runtime")
561                .block_on(connection.close())
562        })
563        .join();
564        if !std::thread::panicking() {
565            closed
566                .expect("join the test writer's teardown")
567                .expect("close the test writer's SQLite connection");
568        }
569    }
570}
571
572#[cfg(test)]
573mod ownership_tests {
574    use super::*;
575
576    #[tokio::test]
577    async fn failed_writer_initialization_releases_sqlite_before_returning_the_schema_error() {
578        let root = crate::test_os::tempdir();
579        let state = root.path().join("state");
580        let repo = root.path().join("repo");
581        std::fs::create_dir(&state).unwrap();
582        std::fs::create_dir(&repo).unwrap();
583        let database = state.join(DATABASE);
584        let connection = rusqlite::Connection::open(&database).unwrap();
585        connection
586            .execute_batch("CREATE TABLE unrelated(value TEXT)")
587            .unwrap();
588        drop(connection);
589        let error = match OwnedReceipts::open(&state, &repo).await {
590            Ok(_) => panic!("an unversioned nonempty database is not our schema"),
591            Err(error) => error,
592        };
593        assert!(error.contains(&schema_error(0)), "{error}");
594        assert!(crate::test_os::database_released(&database).unwrap());
595        root.close().unwrap();
596    }
597
598    #[tokio::test]
599    async fn explicit_writer_close_releases_sqlite_before_return_and_closes_clones() {
600        let root = crate::test_os::tempdir();
601        let state = root.path().join("state");
602        let repo = root.path().join("repo");
603        std::fs::create_dir(&repo).unwrap();
604        let writer = OwnedReceipts::open(&state, &repo).await.unwrap();
605        let retained = Receipts::clone(&writer);
606        let database = state.join(DATABASE);
607        assert!(!crate::test_os::database_released(&database).unwrap());
608        Receipts::clone(&writer).close().await.unwrap();
609        assert!(crate::test_os::database_released(&database).unwrap());
610        assert_eq!(retained.state_id().await.unwrap_err(), "ConnectionClosed");
611        root.close().unwrap();
612    }
613
614    #[tokio::test]
615    async fn owned_writer_closes_before_temp_state_removal_on_success_and_unwind() {
616        for unwind in [false, true] {
617            let root = crate::test_os::tempdir();
618            let state = root.path().join("state");
619            let repo = root.path().join("repo");
620            std::fs::create_dir(&repo).unwrap();
621            let writer = OwnedReceipts::open(&state, &repo).await.unwrap();
622            assert!(writer.state_id().await.is_ok());
623            // Even another writer handle cannot retain the worker after fixture teardown.
624            let retained = Receipts::clone(&writer);
625            if unwind {
626                let failed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
627                    let _writer = writer;
628                    panic!("fixture assertion failure");
629                }));
630                assert!(failed.is_err());
631            } else {
632                drop(writer);
633            }
634            let error = retained.state_id().await.unwrap_err();
635            assert_eq!(error, "ConnectionClosed");
636            root.close().unwrap();
637        }
638    }
639}
640
641/// Notify at the DB-worker completion boundary, even if the awaiting caller was cancelled.
642/// total_changes is conservative (rolled-back changes can invalidate harmlessly). Only a
643/// finished autocommit edge can publish; no event is an audit record or proof of success.
644#[derive(Clone)]
645pub(crate) struct NotifyingConnection {
646    inner: Connection,
647    publisher: crate::changes::Publisher,
648}
649impl NotifyingConnection {
650    pub(super) async fn finish<T>(self, result: Result<T, String>) -> Result<T, String> {
651        super::connection::finish(self.inner, result).await
652    }
653    pub(crate) fn new(inner: Connection, state: &Path) -> Self {
654        Self {
655            inner,
656            publisher: crate::changes::Publisher::new(state),
657        }
658    }
659    pub(crate) async fn call<F, R, E>(&self, function: F) -> Result<R, tokio_rusqlite::Error<E>>
660    where
661        F: FnOnce(&mut rusqlite::Connection) -> Result<R, E> + Send + 'static,
662        R: Send + 'static,
663        E: Send + 'static,
664    {
665        let publisher = self.publisher.clone();
666        self.inner
667            .call(move |db| {
668                let before = db.total_changes();
669                let result = function(db);
670                if db.is_autocommit() && db.total_changes() != before {
671                    publisher.notify(crate::changes::Change::StateInvalidated);
672                }
673                result
674            })
675            .await
676    }
677}
678
679impl Receipts {
680    /// Close the shared SQLite connection before releasing its state files. Any cloned
681    /// receipts also become closed; call this only after all operations on them finish.
682    pub async fn close(self) -> Result<(), String> {
683        self.connection.finish(Ok(())).await
684    }
685
686    pub async fn open(state: &Path, repo: &Path) -> Result<Self, String> {
687        let state = prepare_directory(state, repo).map_err(|e| e.to_string())?;
688        check_files(&state)?;
689        let state_id: crate::types::StateId = crate::agent::uuid()?.parse()?;
690        let connection = Connection::open(state.join(DATABASE))
691            .await
692            .map_err(|e| e.to_string())?;
693        let connection = NotifyingConnection::new(connection, &state);
694        // WAL initialization retries share the database contention budget.
695        let deadline = tokio::time::Instant::now() + super::SQLITE_BUSY_TIMEOUT;
696        loop {
697            let state_id = state_id.clone();
698            let initialized = connection
699                .call(move |db| -> Result<(), Error> {
700                    db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
701                    db.pragma_update(None, "foreign_keys", true)?;
702                    let version: u32 =
703                        db.pragma_query_value(None, "user_version", |row| row.get(0))?;
704                    if version != 0
705                        && !(super::migrations::FIRST_VERSION..=STATE_SCHEMA_VERSION)
706                            .contains(&version)
707                    {
708                        return Err(Error::Invalid(schema_error(version)));
709                    }
710                    db.pragma_update(None, "journal_mode", "WAL")?;
711                    let transaction =
712                        db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
713                    // Publish the schema, the state's id and the version together; readers see
714                    // an empty snapshot until commit.
715                    super::migrations::apply(&transaction)?;
716                    if !schema_initialized(&transaction)? {
717                        transaction.execute_batch(SCHEMA)?;
718                        transaction.execute(
719                            "INSERT INTO state_meta(name,value) VALUES ('id',?)",
720                            [&state_id],
721                        )?;
722                        transaction.pragma_update(None, "user_version", STATE_SCHEMA_VERSION)?;
723                    }
724                    transaction.commit()?;
725                    Ok(())
726                })
727                .await;
728            match initialized {
729                Ok(()) => break,
730                Err(tokio_rusqlite::Error::Error(Error::Sql(rusqlite::Error::SqliteFailure(
731                    error,
732                    _,
733                )))) if error.code == rusqlite::ErrorCode::DatabaseBusy
734                    && tokio::time::Instant::now() < deadline =>
735                {
736                    // Concurrent first opens can race when enabling WAL despite busy_timeout.
737                    tokio::time::sleep(INITIALIZATION_RETRY_INTERVAL).await;
738                }
739                Err(error) => return connection.finish(Err(error.to_string())).await,
740            }
741        }
742        Ok(Self { connection })
743    }
744
745    /// This state's stable id ([`read_state_id`]).
746    pub async fn state_id(&self) -> Result<crate::types::StateId, String> {
747        self.connection
748            .call(|db| -> Result<crate::types::StateId, Error> {
749                Ok(
750                    db.query_row("SELECT value FROM state_meta WHERE name='id'", [], |row| {
751                        row.get(0)
752                    })?,
753                )
754            })
755            .await
756            .map_err(|e| e.to_string())
757    }
758
759    pub async fn create_run(&self, run: &Run, requests: &[Request]) -> Result<(), String> {
760        for request in requests {
761            if request.run_id != run.id {
762                return Err("Request belongs to another Run.".into());
763            }
764        }
765        let run = run.clone();
766        let requests = requests.to_vec();
767        self.connection
768            .call(move |db| -> Result<(), Error> {
769                let transaction = db.transaction()?;
770                transaction.execute(
771                    "INSERT INTO runs(id,repo,status,data,request_key) VALUES (?,?,?,?,?)",
772                    params![
773                        run.id,
774                        crate::platform::path_text(&run.repo_path),
775                        run.status(),
776                        serde_json::to_string(&run)?,
777                        run.request_key
778                    ],
779                )?;
780                for (ordinal, request) in requests.iter().enumerate() {
781                    transaction.execute(
782                        "INSERT INTO requests(id,run_id,eval_id,ordinal,status,data) VALUES (?,?,?,?,?,?)",
783                        params![
784                            request.id,
785                            run.id,
786                            request.eval_id,
787                            ordinal as i64,
788                            request.status(),
789                            serde_json::to_string(request)?
790                        ],
791                    )?;
792                }
793                transaction.commit()?;
794                Ok(())
795            })
796            .await
797            .map_err(|e| e.to_string())
798    }
799
800    pub async fn save_request(&self, request: &Request) -> Result<(), String> {
801        let request = request.clone();
802        self.connection
803            .call(move |db| update_request(db, &request))
804            .await
805            .map_err(|e| e.to_string())
806    }
807
808    pub async fn save_run(&self, run: &Run) -> Result<(), String> {
809        let run = run.clone();
810        self.connection
811            .call(move |db| -> Result<(), Error> {
812                let transaction =
813                    db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
814                super::run_control::save(&transaction, run)?;
815                transaction.commit()?;
816                Ok(())
817            })
818            .await
819            .map_err(|e| e.to_string())
820    }
821
822    pub async fn finish(&self, run: &Run, requests: &[Request]) -> Result<(), String> {
823        let run = run.clone();
824        let requests = requests.to_vec();
825        self.connection
826            .call(move |db| -> Result<(), Error> {
827                let transaction =
828                    db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
829                for request in requests {
830                    // A concurrent submission may already have settled this saved request.
831                    if request.status().is_human_wait() {
832                        continue;
833                    }
834                    transaction.execute("UPDATE requests SET status=?,data=?,execution_id=? WHERE id=? AND status NOT IN ('GREEN','RED','ERROR')", params![request.status(),serde_json::to_string(&request)?,request.execution_id,request.id])?;
835                }
836                super::run_control::save(&transaction, run)?;
837                transaction.commit()?;
838                Ok(())
839            })
840            .await
841            .map_err(|e| e.to_string())
842    }
843}
844
845pub(super) fn update_request(db: &rusqlite::Connection, request: &Request) -> Result<(), Error> {
846    if db.execute(
847        "UPDATE requests SET status=?,data=?,execution_id=? WHERE id=?",
848        params![
849            request.status(),
850            serde_json::to_string(request)?,
851            request.execution_id,
852            request.id
853        ],
854    )? != 1
855    {
856        return Err(Error::Invalid("Review request not found.".into()));
857    }
858    Ok(())
859}
860
861pub(super) fn schema_initialized(transaction: &rusqlite::Transaction<'_>) -> Result<bool, Error> {
862    let version: u32 = transaction.pragma_query_value(None, "user_version", |row| row.get(0))?;
863    // Check both in one snapshot: initialization may commit between separate reads.
864    if version == 0
865        && !transaction.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master)", [], |row| {
866            row.get::<_, bool>(0)
867        })?
868    {
869        return Ok(false);
870    }
871    if version != STATE_SCHEMA_VERSION {
872        return Err(Error::Invalid(schema_error(version)));
873    }
874    Ok(true)
875}
876
877/// Reject non-regular state files and a database of another schema before any read or write.
878pub(super) fn check_files(state: &Path) -> Result<(), String> {
879    match state_schema(state)? {
880        Some(version)
881            if (super::migrations::FIRST_VERSION..STATE_SCHEMA_VERSION).contains(&version) =>
882        {
883            super::migrations::open_existing(state)
884        }
885        Some(version) if version != 0 && version != STATE_SCHEMA_VERSION => {
886            Err(schema_error(version))
887        }
888        _ => Ok(()),
889    }
890}
891
892/// The state database's schema version, read without creating it; `None` without a database.
893/// `doctor` reports it.
894pub fn state_schema(state: &Path) -> Result<Option<u32>, String> {
895    let state = canonical_target(state).map_err(|e| e.to_string())?;
896    regular_files(&state)?;
897    let database = state.join(DATABASE);
898    if !database.try_exists().map_err(|e| e.to_string())? {
899        return Ok(None);
900    }
901    let read = || -> Result<u32, rusqlite::Error> {
902        let db = rusqlite::Connection::open_with_flags(
903            &database,
904            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
905        )?;
906        db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
907        db.pragma_query_value(None, "user_version", |row| row.get(0))
908    };
909    let version = read().map_err(|e| e.to_string())?;
910    if (super::migrations::FIRST_VERSION..STATE_SCHEMA_VERSION).contains(&version) {
911        super::migrations::open_existing(&state)?;
912        return read().map(Some).map_err(|e| e.to_string());
913    }
914    Ok(Some(version))
915}
916
917/// The state's stable id, read without creating anything; `None` until a command that writes
918/// the state has opened it.
919pub fn read_state_id(state: &Path) -> Result<Option<crate::types::StateId>, String> {
920    let state = canonical_target(state).map_err(|e| e.to_string())?;
921    check_files(&state)?;
922    let database = state.join(DATABASE);
923    if !database.try_exists().map_err(|e| e.to_string())? {
924        return Ok(None);
925    }
926    let read = || -> Result<Option<crate::types::StateId>, Error> {
927        let mut db = rusqlite::Connection::open_with_flags(
928            &database,
929            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
930        )?;
931        db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
932        let transaction = db.transaction()?;
933        if !schema_initialized(&transaction)? {
934            return Ok(None);
935        }
936        Ok(transaction
937            .query_row("SELECT value FROM state_meta WHERE name='id'", [], |row| {
938                row.get(0)
939            })
940            .optional()?)
941    };
942    read().map_err(|e| e.to_string())
943}
944
945pub(crate) fn regular_files(state: &Path) -> Result<(), String> {
946    regular_files_with_retry(state, || {
947        std::thread::sleep(super::TRANSIENT_ACCESS_BACKOFF);
948    })
949}
950
951/// The probe and retry path is shared with the seam test; only the between-attempt effect
952/// differs, so a fixture can release a pending-delete handle exactly after a failed probe.
953pub(super) fn regular_files_with_retry(
954    state: &Path,
955    mut before_retry: impl FnMut(),
956) -> Result<(), String> {
957    for suffix in ["", "-wal", "-shm"] {
958        let path = state.join(format!("{DATABASE}{suffix}"));
959        match super::retry_transient_access(|| crate::platform::path_kind(&path), &mut before_retry)
960        {
961            Ok(kind) if kind != crate::platform::FileKind::File => {
962                return Err(format!(
963                    "State files must be regular files: {}",
964                    crate::platform::path_text(&path)
965                ));
966            }
967            Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
968                return Err(error.to_string());
969            }
970            _ => {}
971        }
972    }
973    Ok(())
974}
975
976#[derive(Debug, Serialize)]
977#[serde(rename_all = "camelCase")]
978pub struct LastRequest {
979    pub run_id: RunId,
980    pub verdict: RequestStatus,
981    #[serde(skip_serializing_if = "Option::is_none")]
982    pub fingerprint: Option<Fingerprint>,
983}
984
985/// Last attempts are audit pointers, never evidence for a new current-input query.
986pub async fn read_latest_requests(
987    state: &Path,
988    repo: &Path,
989) -> Result<std::collections::BTreeMap<crate::types::EvalId, LastRequest>, String> {
990    let state = canonical_target(state).map_err(|e| e.to_string())?;
991    outside_workspace(repo, &state).map_err(|e| e.to_string())?;
992    check_files(&state)?;
993    if !state
994        .join(DATABASE)
995        .try_exists()
996        .map_err(|e| e.to_string())?
997    {
998        return Ok(Default::default());
999    }
1000    let connection = Connection::open_with_flags(
1001        state.join(DATABASE),
1002        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
1003    )
1004    .await
1005    .map_err(|e| e.to_string())?;
1006    let repo = repo.to_path_buf();
1007    let result = connection
1008        .call(move |db| -> Result<_, Error> {
1009            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
1010            let transaction = db.transaction()?;
1011            if !schema_initialized(&transaction)? {
1012                return Ok(Default::default());
1013            }
1014            let latest = {
1015                let mut statement = transaction.prepare(
1016                    "SELECT data,run_data FROM (
1017                SELECT q.eval_id,q.data,r.data AS run_data,
1018                    row_number() OVER (PARTITION BY q.eval_id ORDER BY r.rowid DESC) AS rank
1019                FROM requests q JOIN runs r ON r.id=q.run_id
1020                WHERE r.repo=?
1021            ) WHERE rank=1 ORDER BY eval_id",
1022                )?;
1023                statement
1024                    .query_map([crate::platform::path_text(&repo)], |row| {
1025                        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
1026                    })?
1027                    .filter_map(|row| match row {
1028                        Err(error) => Some(Err(error)),
1029                        Ok((data, run_data)) => {
1030                            super::unreadable::evidence::<Run>("run", &run_data)
1031                                .and_then(|_| {
1032                                    super::unreadable::evidence::<Request>("request", &data)
1033                                })
1034                                .map(|request| {
1035                                    Ok((
1036                                        request.eval_id.clone(),
1037                                        LastRequest {
1038                                            run_id: request.run_id.clone(),
1039                                            verdict: request.status(),
1040                                            fingerprint: request.fingerprint,
1041                                        },
1042                                    ))
1043                                })
1044                        }
1045                    })
1046                    .collect::<Result<_, _>>()?
1047            };
1048            transaction.commit()?;
1049            Ok(latest)
1050        })
1051        .await
1052        .map_err(|e| e.to_string());
1053    super::connection::finish(connection, result).await
1054}
1055
1056/// Existing store only: no discovery, schema creation, or execution.
1057pub async fn read_run(state: &Path, id: &str) -> Result<RunView, String> {
1058    let state = canonical_target(state).map_err(|e| e.to_string())?;
1059    check_files(&state)?;
1060    let connection = Connection::open_with_flags(
1061        state.join(DATABASE),
1062        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
1063    )
1064    .await
1065    .map_err(|e| e.to_string())?;
1066    let result = async {
1067        let id: RunId = id.parse()?;
1068        connection
1069            .call(move |db| -> Result<RunView, Error> {
1070                db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
1071                let transaction = db.transaction()?;
1072                if !schema_initialized(&transaction)? {
1073                    return Err(Error::Invalid("Run not found.".into()));
1074                }
1075                let saved: Option<(String, String)> = transaction
1076                    .query_row("SELECT repo,data FROM runs WHERE id=?", [&id], |row| {
1077                        Ok((row.get(0)?, row.get(1)?))
1078                    })
1079                    .optional()?;
1080                let (repo, data) = saved.ok_or_else(|| Error::Invalid("Run not found.".into()))?;
1081                outside_workspace(Path::new(&repo), &state)
1082                    .map_err(|e| Error::Invalid(e.to_string()))?;
1083                let run = super::unreadable::decode("run", &id, &data)
1084                    .map_err(|error| Error::Invalid(error.to_string()))?;
1085                let requests = {
1086                    let mut statement = transaction
1087                        .prepare("SELECT data FROM requests WHERE run_id=? ORDER BY ordinal")?;
1088                    statement
1089                        .query_map([&id], |row| row.get::<_, String>(0))?
1090                        .filter_map(|row| match row {
1091                            Ok(data) => super::unreadable::evidence("request", &data).map(Ok),
1092                            Err(error) => Some(Err(Error::Sql(error))),
1093                        })
1094                        .collect::<Result<Vec<Request>, Error>>()?
1095                };
1096                let unreadable = super::unreadable::in_run(&transaction, &id)?;
1097                transaction.commit()?;
1098                Ok(RunView {
1099                    run,
1100                    requests,
1101                    unreadable,
1102                })
1103            })
1104            .await
1105            .map_err(|e| e.to_string())
1106    }
1107    .await;
1108    // A query result arrives before the DB worker drops its connection. Join the
1109    // close on every return path so callers can immediately release the state files.
1110    super::connection::finish(connection, result).await
1111}
artifactize_core/store/receipts.rs

artifactize_core/store/
receipts.rs
