dpm_store/sqlite/
runs.rs

1//! The run store: a sidecar SQLite file holding agent runs, their durable lifecycle and bounded
2//! activity, apart from the project store.
3//!
4//! Runs are observations, not project facts, so they never enter the semantic operation log or
5//! change the plan revision. Keeping them in their own file leaves the project layout, replay and
6//! writer lock untouched: activity writes contend only with other run writes. The file sits beside
7//! the project store (`<store>.runs`), is bound to one workspace and one lineage when it is
8//! created, and is refused, unchanged, if it is corrupt, of another version, or bound to another
9//! workspace. Reading a store that was never created does not create it.
10
11use crate::{
12    LineageError, RunStoreError, StoreError, StoredRecord,
13    error::database_error,
14    schema::Layout,
15    sqlite::{
16        recovery::files,
17        snapshot::{column, decode},
18    },
19};
20use dpm_model::{LineageId, WorkspaceId};
21use rusqlite::{Connection, OpenFlags, Transaction, TransactionBehavior};
22use serde::{Deserialize, Serialize};
23use std::{
24    path::{Path, PathBuf},
25    time::Duration,
26};
27
28mod activity;
29mod codec;
30pub(super) mod pairing;
31mod read;
32pub(super) mod recovery;
33mod schema;
34mod write;
35
36pub use recovery::RunStoreReport;
37pub use schema::RUN_STORE_VERSION;
38
39/// How long a run write waits for another run writer's lock before reporting the store busy.
40const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
41
42/// Activity records kept by default; older ones are pruned and reported as a gap.
43pub const RUN_ACTIVITY_LIMIT: u64 = 10_000;
44
45/// The workspace and history a run store was created for.
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
47pub struct RunBinding {
48    /// Workspace whose project store this run store sits beside.
49    pub workspace_id: WorkspaceId,
50    /// Lineage of the project store when this run store was created or last restored. Writes are
51    /// refused when the project store continues another lineage.
52    pub lineage_id: LineageId,
53    /// A backup that is read and verified but never written until restored.
54    pub archived: bool,
55}
56
57/// The outcome of a write, and whether it only answered a repeat of an earlier one.
58#[derive(Debug, Clone, PartialEq)]
59pub struct Written<T> {
60    /// What the store recorded.
61    pub value: T,
62    /// `true` when the identity was already recorded with the same content and nothing was
63    /// written.
64    pub replayed: bool,
65}
66
67/// A connection to one run store file.
68pub struct RunStore {
69    connection: Connection,
70    path: PathBuf,
71    activity_limit: u64,
72}
73
74impl RunStore {
75    /// The run store that belongs beside the project store at `project`.
76    #[must_use]
77    pub fn sidecar_path(project: &Path) -> PathBuf {
78        files::side_file(project, ".runs")
79    }
80
81    /// Open an existing run store for the project store of `workspace`, or `None` when none was
82    /// ever created. Nothing is created or written; a store that is corrupt, of another version
83    /// or bound to another workspace is refused before it is opened for writing.
84    pub fn open_existing_blocking(
85        path: &Path,
86        workspace: WorkspaceId,
87    ) -> Result<Option<Self>, RunStoreError> {
88        let exists = path.try_exists().map_err(|source| StoreError::Io {
89            action: "inspect",
90            path: path.to_path_buf(),
91            source,
92        })?;
93        if !exists {
94            return Ok(None);
95        }
96        let probe = files::open_read_only_blocking(path, files::read_mode_blocking(path)?)?;
97        match schema::check_blocking(&probe, path)? {
98            Layout::Empty => return Ok(None),
99            Layout::Current => check_workspace(&read_binding(&probe, path)?, workspace, path)?,
100        }
101        drop(probe);
102        let connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)
103            .map_err(database_error(path, "open run store"))?;
104        Ok(Some(Self::prepare(connection, path.to_path_buf())?))
105    }
106
107    /// Open the run store beside a project store, creating it bound to `workspace` and `lineage`
108    /// when it does not exist yet.
109    pub fn open_or_create_blocking(
110        path: &Path,
111        workspace: WorkspaceId,
112        lineage: LineageId,
113    ) -> Result<Self, RunStoreError> {
114        if let Some(store) = Self::open_existing_blocking(path, workspace)? {
115            return Ok(store);
116        }
117        let connection =
118            Connection::open(path).map_err(database_error(path, "create run store"))?;
119        let mut store = Self::prepare(connection, path.to_path_buf())?;
120        store.initialize_blocking(workspace, lineage)?;
121        Ok(store)
122    }
123
124    /// A disposable run store for an in-memory project store.
125    pub fn in_memory_blocking(
126        workspace: WorkspaceId,
127        lineage: LineageId,
128    ) -> Result<Self, RunStoreError> {
129        let path = PathBuf::from(":memory:");
130        let connection =
131            Connection::open_in_memory().map_err(database_error(&path, "open run store"))?;
132        let mut store = Self::prepare(connection, path)?;
133        store.initialize_blocking(workspace, lineage)?;
134        Ok(store)
135    }
136
137    fn prepare(connection: Connection, path: PathBuf) -> Result<Self, StoreError> {
138        connection
139            .busy_timeout(BUSY_TIMEOUT)
140            .map_err(database_error(&path, "set busy timeout"))?;
141        connection
142            .execute_batch("PRAGMA foreign_keys = ON;")
143            .map_err(database_error(&path, "configure run store"))?;
144        Ok(Self {
145            connection,
146            path,
147            activity_limit: RUN_ACTIVITY_LIMIT,
148        })
149    }
150
151    /// Create the layout in an empty database, unless a concurrent creator already did.
152    fn initialize_blocking(
153        &mut self,
154        workspace: WorkspaceId,
155        lineage: LineageId,
156    ) -> Result<(), RunStoreError> {
157        // The journal mode cannot change inside a transaction.
158        self.connection
159            .execute_batch("PRAGMA journal_mode = WAL;")
160            .map_err(database_error(&self.path, "configure run store"))?;
161        let transaction = self
162            .connection
163            .transaction_with_behavior(TransactionBehavior::Immediate)
164            .map_err(database_error(&self.path, "begin run store creation"))?;
165        if schema::check_blocking(&transaction, &self.path)? == Layout::Empty {
166            schema::create_blocking(&transaction, &self.path)?;
167            transaction
168                .execute(
169                    "INSERT INTO run_binding(singleton, workspace_id, lineage_id, archived) \
170                     VALUES(1, ?1, ?2, 0)",
171                    rusqlite::params![workspace.to_string(), lineage.to_string()],
172                )
173                .map_err(database_error(&self.path, "bind run store"))?;
174            transaction
175                .execute(
176                    "INSERT INTO run_activity_retention(singleton, pruned_through) VALUES(1, 0)",
177                    [],
178                )
179                .map_err(database_error(&self.path, "record retention"))?;
180        }
181        check_workspace(
182            &read_binding(&transaction, &self.path)?,
183            workspace,
184            &self.path,
185        )?;
186        transaction
187            .commit()
188            .map_err(database_error(&self.path, "commit run store creation"))?;
189        Ok(())
190    }
191
192    /// Keep at most `limit` activity records from now on; older ones are pruned by the next write.
193    pub fn set_activity_limit(&mut self, limit: u64) {
194        self.activity_limit = limit;
195    }
196
197    /// Location of the run store, or the in-memory label.
198    #[must_use]
199    pub fn path(&self) -> &Path {
200        &self.path
201    }
202
203    /// The workspace and history this run store is bound to.
204    pub fn binding_blocking(&self) -> Result<RunBinding, RunStoreError> {
205        Ok(read_binding(&self.connection, &self.path)?)
206    }
207
208    fn read_transaction(&self) -> Result<Transaction<'_>, StoreError> {
209        self.connection
210            .unchecked_transaction()
211            .map_err(database_error(&self.path, "begin run query"))
212    }
213}
214
215fn check_workspace(
216    binding: &RunBinding,
217    expected: WorkspaceId,
218    path: &Path,
219) -> Result<(), RunStoreError> {
220    if binding.workspace_id == expected {
221        return Ok(());
222    }
223    Err(RunStoreError::WorkspaceMismatch {
224        path: path.to_path_buf(),
225        expected,
226        actual: binding.workspace_id,
227    })
228}
229
230pub(super) fn read_binding(connection: &Connection, path: &Path) -> Result<RunBinding, StoreError> {
231    let mut statement = connection
232        .prepare("SELECT workspace_id, lineage_id, archived FROM run_binding WHERE singleton = 1")
233        .map_err(database_error(path, "prepare run binding query"))?;
234    let mut rows = statement
235        .query([])
236        .map_err(database_error(path, "load run binding"))?;
237    let Some(row) = rows
238        .next()
239        .map_err(database_error(path, "load run binding"))?
240    else {
241        return Err(StoreError::Lineage(LineageError::Missing {
242            path: path.to_path_buf(),
243        }));
244    };
245    let scalar = |index, field| -> Result<String, StoreError> {
246        column::<String>(row, index, path, StoredRecord::Run, field)
247            .map(|text| serde_json::Value::String(text).to_string())
248    };
249    Ok(RunBinding {
250        workspace_id: decode(
251            &scalar(0, "workspace_id")?,
252            path,
253            StoredRecord::Run,
254            "workspace_id",
255        )?,
256        lineage_id: decode(
257            &scalar(1, "lineage_id")?,
258            path,
259            StoredRecord::Run,
260            "lineage_id",
261        )?,
262        archived: column(row, 2, path, StoredRecord::Run, "archived")?,
263    })
264}
265
266#[cfg(test)]
267mod tests;
dpm_store/sqlite/runs.rs

dpm_store/sqlite/
runs.rs
