uqa_storage/
backend.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Persistent storage backend factory.
8//!
9//! This boundary keeps the engine from constructing SQLite-backed stores
10//! directly. Alternative persistent backends can implement the same factory
11//! without changing query execution code.
12
13use std::collections::BTreeMap;
14use std::path::{Path, PathBuf};
15use std::sync::atomic::{AtomicU64, Ordering};
16use std::sync::Arc;
17
18use uqa_analysis::Analyzer;
19use uqa_core::{DocId, Value};
20
21use crate::document_store::DocumentStore;
22use crate::inverted_index::InvertedIndex;
23use crate::vector_index::{VectorIndex, VectorIndexOpenMode, VectorIndexSpec};
24use crate::CatalogFacade;
25
26#[derive(Debug, thiserror::Error)]
27pub enum StorageBackendError {
28    #[error(transparent)]
29    Memory(#[from] uqa_core::memory::MemoryError),
30    #[error(transparent)]
31    Cancelled(#[from] uqa_core::QueryCancelled),
32    #[error("text analysis failed: {0}")]
33    Analysis(#[from] uqa_analysis::AnalysisError),
34    #[error("payload serialization failed: {0}")]
35    Serde(#[from] serde_json::Error),
36    #[error("{backend} storage failed: {source}")]
37    Backend {
38        backend: &'static str,
39        #[source]
40        source: Box<dyn std::error::Error + Send + Sync>,
41    },
42    #[error("{0}")]
43    Other(String),
44}
45
46impl StorageBackendError {
47    pub fn backend(
48        backend: &'static str,
49        source: impl std::error::Error + Send + Sync + 'static,
50    ) -> Self {
51        Self::Backend {
52            backend,
53            source: Box::new(source),
54        }
55    }
56}
57
58pub type StorageBackendResult<T> = std::result::Result<T, StorageBackendError>;
59
60/// Opaque transaction checkpoint identity. SQL savepoint names remain engine metadata and are never forwarded into a backend namespace.
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
62pub struct StorageSavepointId(u64);
63
64impl StorageSavepointId {
65    #[must_use]
66    pub fn allocate() -> Self {
67        static NEXT_ID: AtomicU64 = AtomicU64::new(1);
68        let id = NEXT_ID
69            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
70                current.checked_add(1)
71            })
72            .expect("storage savepoint identity space exhausted");
73        Self(id)
74    }
75
76    pub fn backend_name(self) -> String {
77        self.0.to_string()
78    }
79}
80
81/// Session-bound catalog and physical storage handles created together.
82///
83/// Both handles must share the same transaction context. Keeping their
84/// construction behind one provider prevents a catalog write from escaping
85/// through a different connection or transaction than document/index writes.
86pub struct PersistentStorageSession {
87    pub catalog: Arc<dyn CatalogFacade>,
88    pub backend: Arc<dyn PersistentStorageBackend>,
89}
90
91/// Stable identity of one durable database. File identities allow engine coordination to extend across independently constructed providers and OS processes; opaque identities coordinate providers inside one process.
92#[derive(Clone, Debug, PartialEq, Eq, Hash)]
93pub enum PersistentStorageIdentity {
94    File(PathBuf),
95    Opaque(String),
96}
97
98impl PersistentStorageIdentity {
99    /// Resolve the stable file identity for a database path. The database file itself may not exist yet because backends materialize it on first write, so a missing file anchors the identity on its canonicalized parent directory instead of failing.
100    pub fn for_database_path(path: &Path) -> StorageBackendResult<Self> {
101        let path = resolve_final_symlinks(path)?;
102        match std::fs::canonicalize(&path) {
103            Ok(canonical) => Ok(Self::File(canonical)),
104            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
105                let file_name = path.file_name().ok_or_else(|| {
106                    StorageBackendError::Other(format!(
107                        "database path `{}` has no file name",
108                        path.display()
109                    ))
110                })?;
111                let parent = path
112                    .parent()
113                    .filter(|parent| !parent.as_os_str().is_empty())
114                    .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
115                let parent = std::fs::canonicalize(&parent).map_err(|error| {
116                    StorageBackendError::Other(format!(
117                        "canonicalize database directory `{}`: {error}",
118                        parent.display()
119                    ))
120                })?;
121                Ok(Self::File(parent.join(file_name)))
122            }
123            Err(error) => Err(StorageBackendError::Other(format!(
124                "canonicalize database `{}`: {error}",
125                path.display()
126            ))),
127        }
128    }
129}
130
131fn resolve_final_symlinks(path: &Path) -> StorageBackendResult<PathBuf> {
132    let mut current = path.to_path_buf();
133    let mut followed = 0usize;
134    loop {
135        match std::fs::symlink_metadata(&current) {
136            Ok(metadata) if metadata.file_type().is_symlink() => {
137                followed += 1;
138                if followed > 40 {
139                    return Err(StorageBackendError::Other(format!(
140                        "database path `{}` has too many symbolic-link levels",
141                        path.display()
142                    )));
143                }
144                let target = std::fs::read_link(&current).map_err(|error| {
145                    StorageBackendError::Other(format!(
146                        "read database symbolic link `{}`: {error}",
147                        current.display()
148                    ))
149                })?;
150                current = if target.is_absolute() {
151                    target
152                } else {
153                    current
154                        .parent()
155                        .filter(|parent| !parent.as_os_str().is_empty())
156                        .unwrap_or_else(|| Path::new("."))
157                        .join(target)
158                };
159            }
160            Ok(_) => return Ok(current),
161            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(current),
162            Err(error) => {
163                return Err(StorageBackendError::Other(format!(
164                    "inspect database path `{}`: {error}",
165                    current.display()
166                )))
167            }
168        }
169    }
170}
171
172impl PersistentStorageSession {
173    pub fn new(
174        catalog: Arc<dyn CatalogFacade>,
175        backend: Arc<dyn PersistentStorageBackend>,
176    ) -> Self {
177        Self { catalog, backend }
178    }
179}
180
181/// Factory for independent sessions over one durable database.
182///
183/// A provider owns the database-level resource while each returned session
184/// owns its transaction state. This is the engine-facing extension point for
185/// `SQLite`, redb, and application-defined Key/Value stores.
186pub trait PersistentStorageProvider: Send + Sync {
187    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession>;
188
189    /// Open handles for initial Engine restoration. Providers may defer catalog schema preparation until `CatalogFacade::initialize_storage` runs inside the owning transaction; ordinary session factories must return an initialized catalog.
190    fn open_initial_session(&self) -> StorageBackendResult<PersistentStorageSession> {
191        self.open_session()
192    }
193
194    /// Return the database identity shared by every session this provider opens. Custom providers that cannot expose a stable identity may keep the default; engines built from the same `Arc` provider still share an in-process coordinator.
195    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
196        Ok(None)
197    }
198
199    /// Protect engine-owned auxiliary files with the database's encryption
200    /// credential. Encrypted file providers must forward this capability;
201    /// `None` declares that their auxiliary storage may be unencrypted.
202    fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
203        None
204    }
205}
206
207/// Factory plus transaction surface for persistent table/index storage.
208pub trait PersistentStorageBackend: Send + Sync {
209    /// Return the stable database identity for independently constructed engines over this backend. File identities also enable cross-process row-lock coordination.
210    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
211        Ok(None)
212    }
213
214    /// Return the encryption credential for database-owned auxiliary files,
215    /// including engines constructed directly from this backend.
216    fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
217        None
218    }
219
220    /// Open a transaction-isolated catalog/backend pair over the same durable database. Engines constructed from already-open backends retain this factory so a row-lock recheck can read the latest committed tuple while the caller's statement snapshot remains pinned.
221    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
222        Err(StorageBackendError::Other(
223            "independent sessions are not implemented for this persistent backend".into(),
224        ))
225    }
226
227    /// Whether an independently pinned read session can remain open while another session writes the same database. Backends that return `false` use a detached fixed snapshot for `REPEATABLE READ` and `SERIALIZABLE` transactions.
228    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
229        false
230    }
231
232    fn document_store(&self, table: &str) -> Box<dyn DocumentStore>;
233
234    /// Upgrade backend-owned document records before table handles are restored. Implementations must make the rewrite atomic and idempotent.
235    fn migrate_document_storage(&self) -> StorageBackendResult<()> {
236        Ok(())
237    }
238
239    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex>;
240
241    /// Upgrade backend-owned inverted-index values before table handles are
242    /// restored. Implementations must make the rewrite atomic and idempotent.
243    fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
244        Ok(())
245    }
246
247    fn vector_index(
248        &self,
249        table: &str,
250        field: &str,
251        dimensions: u32,
252        spec: VectorIndexSpec,
253        mode: VectorIndexOpenMode,
254    ) -> StorageBackendResult<Box<dyn VectorIndex>>;
255
256    fn drop_vector_index_metadata(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
257        Ok(())
258    }
259
260    /// Whether this backend maps logical `btree` indexes to durable postings.
261    fn persists_btree_indexes(&self) -> bool {
262        false
263    }
264
265    /// `Some(entries)` is a complete persisted index; `None` means it has not
266    /// been built yet and the engine must backfill it from documents once.
267    fn load_btree_index(
268        &self,
269        _table: &str,
270        _field: &crate::ValueIndexKey,
271    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
272        Ok(None)
273    }
274
275    fn btree_index_fields(&self, _table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
276        Ok(Vec::new())
277    }
278
279    /// Fields whose persisted posting support was found inconsistent during a
280    /// schema migration. The engine repairs these at its explicit open-time
281    /// write boundary and clears each durable retry marker only after success.
282    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, crate::ValueIndexKey)>> {
283        Ok(Vec::new())
284    }
285
286    fn clear_btree_index_repair(
287        &self,
288        _table: &str,
289        _field: &crate::ValueIndexKey,
290    ) -> StorageBackendResult<()> {
291        Ok(())
292    }
293
294    fn replace_btree_index(
295        &self,
296        _table: &str,
297        _field: &crate::ValueIndexKey,
298        _values: &[(DocId, Value)],
299    ) -> StorageBackendResult<()> {
300        Ok(())
301    }
302
303    /// Repair sparse support differences without requiring capable backends
304    /// to rewrite every already-valid posting. The complete replacement is
305    /// supplied for the storage-neutral fallback.
306    fn repair_btree_index(
307        &self,
308        table: &str,
309        field: &crate::ValueIndexKey,
310        complete: &[(DocId, Value)],
311        _stale_doc_ids: &[DocId],
312        _missing: &[(DocId, Value)],
313    ) -> StorageBackendResult<()> {
314        self.replace_btree_index(table, field, complete)
315    }
316
317    /// Replace several complete indexes for one table atomically. Backends
318    /// may override this to share one transaction and prepared statements.
319    fn replace_btree_indexes(
320        &self,
321        table: &str,
322        indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
323    ) -> StorageBackendResult<()> {
324        for (field, values) in indexes {
325            self.replace_btree_index(table, field, values)?;
326        }
327        Ok(())
328    }
329
330    fn apply_btree_index_write(
331        &self,
332        _table: &str,
333        _doc_id: DocId,
334        _values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
335    ) -> StorageBackendResult<()> {
336        Ok(())
337    }
338
339    fn drop_btree_index(
340        &self,
341        _table: &str,
342        _field: &crate::ValueIndexKey,
343    ) -> StorageBackendResult<()> {
344        Ok(())
345    }
346
347    fn clear_btree_indexes(&self, _table: &str) -> StorageBackendResult<()> {
348        Ok(())
349    }
350
351    /// Reclaim backend-owned storage outside a transaction. Backends whose logical stores eagerly remove obsolete values may keep the no-op default; durable backends with file-level compaction should override it.
352    fn vacuum(&self) -> StorageBackendResult<()> {
353        Ok(())
354    }
355
356    fn begin_transaction(&self) -> StorageBackendResult<()>;
357
358    /// Begin a transaction whose first operation is expected to be a read.
359    /// Backends with distinct lock modes may defer write-lock acquisition;
360    /// the default preserves existing transaction semantics.
361    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
362        self.begin_transaction()
363    }
364
365    /// Begin an atomic transaction that may remain read-only or perform writes after its initial reads. Backends without read-to-write promotion acquire a writer transaction immediately.
366    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
367        self.begin_transaction()
368    }
369
370    /// Whether this session currently owns a pinned storage transaction.
371    fn in_transaction(&self) -> bool;
372
373    /// Whether the current transaction has performed a physical write.
374    ///
375    /// The engine uses this to enforce read-only statement transactions even
376    /// for writes made through catalog/index helpers it did not classify.
377    fn transaction_has_written(&self) -> StorageBackendResult<bool>;
378
379    /// Backend commit generation visible to this session, when available.
380    /// A changing value invalidates session-local catalog and index caches.
381    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
382        Ok(None)
383    }
384
385    /// Whether reading [`Self::change_version`] can proceed while this session owns its pinned transaction. An independent monitor can also be unsafe for a reader when a pending writer is waiting for that reader's lock.
386    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
387        Ok(true)
388    }
389
390    /// Pin the transaction's read snapshot before cache restoration.
391    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
392        Ok(())
393    }
394
395    fn commit_transaction(&self) -> StorageBackendResult<()>;
396
397    fn rollback_transaction(&self) -> StorageBackendResult<()>;
398
399    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
400
401    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
402
403    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
404}
405
406#[cfg(test)]
407mod identity_tests {
408    use super::*;
409
410    #[cfg(unix)]
411    #[test]
412    fn dangling_database_symlink_keeps_the_target_identity_after_creation() {
413        use std::os::unix::fs::symlink;
414
415        let directory = tempfile::tempdir().unwrap();
416        let target = directory.path().join("target.db");
417        let link = directory.path().join("database.db");
418        symlink("target.db", &link).unwrap();
419
420        let before = PersistentStorageIdentity::for_database_path(&link).unwrap();
421        std::fs::File::create(&target).unwrap();
422        let after = PersistentStorageIdentity::for_database_path(&link).unwrap();
423
424        assert_eq!(before, after);
425        assert_eq!(
426            after,
427            PersistentStorageIdentity::File(target.canonicalize().unwrap())
428        );
429    }
430}
uqa_storage/backend.rs

uqa_storage/
backend.rs
