dpm_store/
sqlite.rs

1use crate::StoreLineage;
2use crate::{
3    StoreError,
4    error::database_error,
5    schema::{self, Layout},
6};
7use dpm_engine::{Operation, apply_command};
8use dpm_model::{LineageId, Plan};
9use rusqlite::{Connection, OpenFlags, TransactionBehavior};
10use std::{
11    path::{Path, PathBuf},
12    time::Duration,
13};
14
15mod history;
16mod integrity;
17mod lineage;
18mod recovery;
19use recovery::files;
20mod replay;
21mod runs;
22pub(crate) use crate::{HistoryEntry, RecordedOperation};
23pub use recovery::{IntegrityReport, restore_store_blocking, verify_store_blocking};
24pub use runs::{
25    RUN_ACTIVITY_LIMIT, RUN_STORE_VERSION, RunBinding, RunStore, RunStoreReport, Written,
26};
27mod snapshot;
28mod snapshot_cache;
29use snapshot::load_blocking;
30
31/// How long a write waits for another connection's lock before reporting the store busy.
32const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
33
34/// A synchronous local database with validated snapshots and immutable operation history.
35pub struct SqliteStore {
36    connection: Connection,
37    cache: snapshot_cache::SnapshotCache,
38    path: PathBuf,
39}
40
41impl SqliteStore {
42    /// Create or open a database without writing to it; [`Self::initialize_blocking`] creates the
43    /// schema.
44    pub fn open_blocking(path: impl AsRef<Path>) -> Result<Self, StoreError> {
45        let path = path.as_ref().to_path_buf();
46        check_existing_read_only_blocking(&path)?;
47        let connection = Connection::open(&path).map_err(database_error(&path, "open database"))?;
48        Self::prepare_blocking(connection, path)
49    }
50
51    /// Open an existing database without creating a new file or writing to it; a store in a
52    /// retired or newer layout is refused unchanged.
53    pub fn open_existing_blocking(path: impl AsRef<Path>) -> Result<Self, StoreError> {
54        let path = path.as_ref().to_path_buf();
55        check_existing_read_only_blocking(&path)?;
56        let connection = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_WRITE)
57            .map_err(database_error(&path, "open existing database"))?;
58        Self::prepare_blocking(connection, path)
59    }
60
61    /// Create an isolated in-memory database for ephemeral work and tests.
62    pub fn in_memory_blocking() -> Result<Self, StoreError> {
63        let path = PathBuf::from(":memory:");
64        let connection =
65            Connection::open_in_memory().map_err(database_error(&path, "open database"))?;
66        Self::prepare_blocking(connection, path)
67    }
68
69    fn prepare_blocking(connection: Connection, path: PathBuf) -> Result<Self, StoreError> {
70        connection
71            .busy_timeout(BUSY_TIMEOUT)
72            .map_err(database_error(&path, "set busy timeout"))?;
73        // Refused formats are checked before this read-write connection opens; schema creation
74        // happens inside the initializing write transaction.
75        schema::check_blocking(&connection, &path)?;
76        connection
77            .execute_batch("PRAGMA foreign_keys = ON;")
78            .map_err(database_error(&path, "configure database"))?;
79        Ok(Self {
80            connection,
81            path,
82            cache: Default::default(),
83        })
84    }
85
86    /// Database location, or the in-memory label.
87    pub fn path(&self) -> &Path {
88        &self.path
89    }
90
91    /// Wait at most `timeout` for another connection's lock before failing as busy.
92    pub fn set_busy_timeout_blocking(&self, timeout: Duration) -> Result<(), StoreError> {
93        self.connection
94            .busy_timeout(timeout)
95            .map_err(database_error(&self.path, "set busy timeout"))
96    }
97
98    /// The lineage this file continues, if initialized.
99    pub fn lineage_blocking(&self) -> Result<Option<StoreLineage>, StoreError> {
100        if schema::check_blocking(&self.connection, &self.path)? == Layout::Empty {
101            return Ok(None);
102        }
103        lineage::read_blocking(&self.connection, &self.path).map(Some)
104    }
105
106    /// The committed revision and its lineage, read in one transaction without the snapshot text,
107    /// so polling for changes costs two row reads; `None` for an uninitialized store.
108    pub fn revision_blocking(&self) -> Result<Option<crate::StoreRevision>, StoreError> {
109        read_revision_blocking(&self.connection, &self.path)
110    }
111
112    /// The plan as it stood before the operation at local sequence `sequence`, replayed from the
113    /// genesis plan; a resent plan change is compared against the state it was first applied to.
114    pub fn plan_before_blocking(&self, sequence: u64) -> Result<Plan, StoreError> {
115        let transaction = self
116            .connection
117            .unchecked_transaction()
118            .map_err(database_error(&self.path, "begin replay"))?;
119        let genesis = snapshot::genesis_blocking(&transaction, &self.path)?.ok_or_else(|| {
120            StoreError::MissingGenesis {
121                path: self.path.clone(),
122            }
123        })?;
124        let plan = replay::replay_before_blocking(&transaction, &self.path, genesis, sequence)?;
125        transaction
126            .commit()
127            .map_err(database_error(&self.path, "finish replay"))?;
128        Ok(plan)
129    }
130
131    /// Read and validate the authoritative snapshot, if initialized.
132    pub fn load_blocking(&self) -> Result<Option<Plan>, StoreError> {
133        if schema::check_blocking(&self.connection, &self.path)? == Layout::Empty {
134            return Ok(None);
135        }
136        snapshot::load_cached_blocking(&self.connection, &self.path, &self.cache)
137    }
138
139    /// Initialize an empty database with `plan` as both the snapshot and the genesis plan the
140    /// history replays from; an existing snapshot or operation history is never erased, and a
141    /// refused initialization writes nothing.
142    pub fn initialize_blocking(&mut self, plan: &Plan) -> Result<(), StoreError> {
143        plan.validate()?;
144        if occupied_blocking(&self.connection, &self.path)? {
145            return Err(StoreError::AlreadyInitialized(self.path.clone()));
146        }
147        // The journal mode cannot change inside a transaction.
148        self.connection
149            .execute_batch("PRAGMA journal_mode = WAL;")
150            .map_err(database_error(&self.path, "configure database"))?;
151        let transaction = self
152            .connection
153            .transaction_with_behavior(TransactionBehavior::Immediate)
154            .map_err(database_error(&self.path, "begin initialization"))?;
155        // A concurrent creator may have committed between the unlocked check and this lock.
156        prepare_write_blocking(&transaction, &self.path)?;
157        if occupied_blocking(&transaction, &self.path)? {
158            return Err(StoreError::AlreadyInitialized(self.path.clone()));
159        }
160        snapshot::write_initial_blocking(&transaction, &self.path, plan)?;
161        transaction
162            .commit()
163            .map_err(database_error(&self.path, "commit initialization"))?;
164        Ok(())
165    }
166
167    /// Atomically append one valid operation and the exact snapshot produced by that command,
168    /// under this file's lineage.
169    ///
170    /// Inside the write transaction, an identity that is already recorded is answered with
171    /// [`StoreError::DuplicateOperation`] before any precondition, an archive is refused, and so
172    /// is a lineage other than `expected_lineage` or a revision other than the operation's base.
173    pub fn persist_blocking(
174        &mut self,
175        plan: &Plan,
176        operation: &Operation,
177        expected_lineage: Option<LineageId>,
178    ) -> Result<HistoryEntry, StoreError> {
179        if operation.base_revision.wrapping_add(1) != operation.resulting_revision
180            || plan.revision != operation.resulting_revision
181        {
182            return Err(StoreError::InvalidOperationRevision {
183                base: operation.base_revision,
184                result: operation.resulting_revision,
185                plan: plan.revision,
186            });
187        }
188        plan.validate()?;
189        let transaction = self
190            .connection
191            .transaction_with_behavior(TransactionBehavior::Immediate)
192            .map_err(database_error(&self.path, "begin operation"))?;
193        if prepare_write_blocking(&transaction, &self.path)? == Layout::Empty {
194            return Err(StoreError::NotInitialized(self.path.clone()));
195        }
196        if let Some(recorded) = history::recorded_blocking(&transaction, &self.path, operation.id)?
197        {
198            return Err(StoreError::DuplicateOperation {
199                recorded: Box::new(recorded),
200            });
201        }
202        let found = lineage::read_blocking(&transaction, &self.path)?;
203        found.check_writable(expected_lineage, &self.path)?;
204        let mut expected = snapshot::load_cached_blocking(&transaction, &self.path, &self.cache)?
205            .ok_or_else(|| StoreError::NotInitialized(self.path.clone()))?;
206        if expected.revision != operation.base_revision {
207            return Err(StoreError::RevisionConflict {
208                expected: operation.base_revision,
209                actual: expected.revision,
210            });
211        }
212        apply_command(
213            &mut expected,
214            operation.actor.clone(),
215            operation.command.clone(),
216            operation.timestamp,
217            operation.id,
218        )?;
219        if expected != *plan {
220            return Err(StoreError::SnapshotMismatch(operation.id));
221        }
222        let sequence = snapshot::write_operation_blocking(
223            &transaction,
224            &self.path,
225            plan,
226            operation,
227            found.lineage_id,
228        )?;
229        transaction
230            .commit()
231            .map_err(database_error(&self.path, "commit operation"))?;
232        Ok(HistoryEntry {
233            sequence,
234            operation: RecordedOperation {
235                operation: operation.clone(),
236                workspace_id: plan.workspace.id,
237                lineage_id: found.lineage_id,
238            },
239        })
240    }
241
242    /// Count committed semantic operations.
243    pub fn operation_count_blocking(&self) -> Result<u64, StoreError> {
244        if schema::check_blocking(&self.connection, &self.path)? == Layout::Empty {
245            return Ok(0);
246        }
247        let count: u64 = self
248            .connection
249            .query_row("SELECT COUNT(*) FROM operations", [], |row| row.get(0))
250            .map_err(database_error(&self.path, "count operations"))?;
251        Ok(count)
252    }
253}
254
255/// Check the version and layout of an existing file through a read-only connection first.
256///
257/// Closing the last read-write connection to a WAL store checkpoints the log into the main file and
258/// deletes the side files, so a store this binary refuses must never be opened read-write at all.
259/// Revision and lineage of the store at `path` through a read-only connection, without decoding
260/// its snapshot, for a client polling a store it has not opened (a repointable project locator);
261/// `None` for an uninitialized store. Stores in another layout are refused as by an open.
262pub fn store_revision_blocking(path: &Path) -> Result<Option<crate::StoreRevision>, StoreError> {
263    let mode = files::read_mode_blocking(path)?;
264    let connection = files::open_read_only_blocking(path, mode)?;
265    read_revision_blocking(&connection, path)
266}
267
268fn read_revision_blocking(
269    connection: &Connection,
270    path: &Path,
271) -> Result<Option<crate::StoreRevision>, StoreError> {
272    if schema::check_blocking(connection, path)? == Layout::Empty {
273        return Ok(None);
274    }
275    let transaction = connection
276        .unchecked_transaction()
277        .map_err(database_error(path, "begin revision query"))?;
278    let revision = snapshot::revision_blocking(&transaction, path)?;
279    let lineage = lineage::read_blocking(&transaction, path)?;
280    let history_head: u64 = transaction
281        .query_row(
282            "SELECT COALESCE(MAX(sequence), 0) FROM operations",
283            [],
284            |row| row.get(0),
285        )
286        .map_err(database_error(path, "read history head"))?;
287    transaction
288        .commit()
289        .map_err(database_error(path, "finish revision query"))?;
290    Ok(revision.map(|revision| crate::StoreRevision {
291        revision,
292        lineage,
293        history_head,
294    }))
295}
296
297fn check_existing_read_only_blocking(path: &Path) -> Result<(), StoreError> {
298    let exists = path.try_exists().map_err(|source| StoreError::Io {
299        action: "inspect",
300        path: path.to_path_buf(),
301        source,
302    })?;
303    if !exists {
304        return Ok(());
305    }
306    let mode = files::read_mode_blocking(path)?;
307    let connection = files::open_read_only_blocking(path, mode)?;
308    if schema::check_blocking(&connection, path)? == Layout::Current {
309        snapshot::load_blocking(&connection, path)?;
310        snapshot::genesis_blocking(&connection, path)?;
311        lineage::read_blocking(&connection, path)?;
312    }
313    Ok(())
314}
315
316fn occupied_blocking(connection: &Connection, path: &Path) -> Result<bool, StoreError> {
317    if schema::check_blocking(connection, path)? == Layout::Empty {
318        return Ok(false);
319    }
320    connection
321        .query_row(
322            "SELECT EXISTS(SELECT 1 FROM plan_state) OR EXISTS(SELECT 1 FROM operations)",
323            [],
324            |row| row.get(0),
325        )
326        .map_err(database_error(path, "check initialization"))
327}
328
329/// Create the schema of an empty store inside a caller-owned write transaction and return the
330/// layout found before that.
331///
332/// The exact layout is re-checked under the write lock, so a schema object added by another
333/// process after open never runs inside this write.
334fn prepare_write_blocking(connection: &Connection, path: &Path) -> Result<Layout, StoreError> {
335    let found = schema::check_blocking(connection, path)?;
336    if found == Layout::Empty {
337        schema::create_blocking(connection, path)?;
338    }
339    Ok(found)
340}
341
342#[cfg(test)]
343mod tests;
dpm_store/sqlite.rs

dpm_store/
sqlite.rs
