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/// Process-local identity of one storage transaction context. Clone it for handles sharing that context; allocate a new identity for each independent session. This is not a durable database or transaction ID.
61#[derive(Clone, Debug)]
62pub struct StorageSessionAffinity(Arc<()>);
63
64impl StorageSessionAffinity {
65    #[must_use]
66    pub fn new() -> Self {
67        Self(Arc::new(()))
68    }
69}
70
71impl Default for StorageSessionAffinity {
72    fn default() -> Self {
73        Self::new()
74    }
75}
76
77impl PartialEq for StorageSessionAffinity {
78    fn eq(&self, other: &Self) -> bool {
79        Arc::ptr_eq(&self.0, &other.0)
80    }
81}
82
83impl Eq for StorageSessionAffinity {}
84
85#[derive(Debug, thiserror::Error)]
86#[error("catalog and data backend must share one storage transaction context")]
87pub struct StorageSessionMismatch;
88
89/// Opaque transaction checkpoint identity. SQL savepoint names remain engine metadata and are never forwarded into a backend namespace.
90#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
91pub struct StorageSavepointId(u64);
92
93impl StorageSavepointId {
94    #[must_use]
95    pub fn allocate() -> Self {
96        static NEXT_ID: AtomicU64 = AtomicU64::new(1);
97        let id = NEXT_ID
98            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
99                current.checked_add(1)
100            })
101            .expect("storage savepoint identity space exhausted");
102        Self(id)
103    }
104
105    pub fn backend_name(self) -> String {
106        self.0.to_string()
107    }
108}
109
110/// Session-bound catalog and physical storage handles created together.
111///
112/// Both handles must share the same transaction context. Keeping their
113/// construction behind one provider prevents a catalog write from escaping
114/// through a different connection or transaction than document/index writes.
115pub struct PersistentStorageSession {
116    pub catalog: Arc<dyn CatalogFacade>,
117    pub backend: Arc<dyn PersistentStorageBackend>,
118}
119
120/// 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.
121#[derive(Clone, Debug, PartialEq, Eq, Hash)]
122pub enum PersistentStorageIdentity {
123    File(PathBuf),
124    Opaque(String),
125}
126
127impl PersistentStorageIdentity {
128    /// 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.
129    pub fn for_database_path(path: &Path) -> StorageBackendResult<Self> {
130        let path = resolve_final_symlinks(path)?;
131        match std::fs::canonicalize(&path) {
132            Ok(canonical) => Ok(Self::File(canonical)),
133            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
134                let file_name = path.file_name().ok_or_else(|| {
135                    StorageBackendError::Other(format!(
136                        "database path `{}` has no file name",
137                        path.display()
138                    ))
139                })?;
140                let parent = path
141                    .parent()
142                    .filter(|parent| !parent.as_os_str().is_empty())
143                    .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
144                let parent = std::fs::canonicalize(&parent).map_err(|error| {
145                    StorageBackendError::Other(format!(
146                        "canonicalize database directory `{}`: {error}",
147                        parent.display()
148                    ))
149                })?;
150                Ok(Self::File(parent.join(file_name)))
151            }
152            Err(error) => Err(StorageBackendError::Other(format!(
153                "canonicalize database `{}`: {error}",
154                path.display()
155            ))),
156        }
157    }
158}
159
160fn resolve_final_symlinks(path: &Path) -> StorageBackendResult<PathBuf> {
161    let mut current = path.to_path_buf();
162    let mut followed = 0usize;
163    loop {
164        match std::fs::symlink_metadata(&current) {
165            Ok(metadata) if metadata.file_type().is_symlink() => {
166                followed += 1;
167                if followed > 40 {
168                    return Err(StorageBackendError::Other(format!(
169                        "database path `{}` has too many symbolic-link levels",
170                        path.display()
171                    )));
172                }
173                let target = std::fs::read_link(&current).map_err(|error| {
174                    StorageBackendError::Other(format!(
175                        "read database symbolic link `{}`: {error}",
176                        current.display()
177                    ))
178                })?;
179                current = if target.is_absolute() {
180                    target
181                } else {
182                    current
183                        .parent()
184                        .filter(|parent| !parent.as_os_str().is_empty())
185                        .unwrap_or_else(|| Path::new("."))
186                        .join(target)
187                };
188            }
189            Ok(_) => return Ok(current),
190            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(current),
191            Err(error) => {
192                return Err(StorageBackendError::Other(format!(
193                    "inspect database path `{}`: {error}",
194                    current.display()
195                )))
196            }
197        }
198    }
199}
200
201/// How a catalog/backend pair retains writes between SQL commands. This contract is independent of the physical database's single-writer commit admission.
202#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
203pub enum StorageTransactionModel {
204    #[default]
205    ProviderSerialized,
206    /// Private writes, retained readers and savepoints share this database incarnation. Command refresh must preserve evaluated writes and commit must validate their original revisions.
207    VersionedConcurrent { database: crate::mvcc::DatabaseId },
208}
209
210impl StorageTransactionModel {
211    pub fn is_versioned(self) -> bool {
212        matches!(self, Self::VersionedConcurrent { .. })
213    }
214}
215
216impl PersistentStorageSession {
217    pub fn new(
218        catalog: Arc<dyn CatalogFacade>,
219        backend: Arc<dyn PersistentStorageBackend>,
220    ) -> Self {
221        Self { catalog, backend }
222    }
223
224    /// Check the transaction model, database incarnation and session affinity before restoration, migration or attachment. Versioned pairs require a matching reported affinity and backend write cancellation; serialized pairs may omit both capabilities.
225    pub fn validate_transaction_affinity(&self) -> StorageBackendResult<()> {
226        let affinity = self.backend.transaction_affinity();
227        let model = self.backend.transaction_model();
228        if self.catalog.transaction_affinity() != affinity
229            || self.catalog.transaction_model() != model
230            || (model.is_versioned() && affinity.is_none())
231        {
232            return Err(StorageBackendError::backend(
233                "session",
234                StorageSessionMismatch,
235            ));
236        }
237        if model.is_versioned() && self.backend.write_cancellation().is_none() {
238            return Err(StorageBackendError::Other(
239                "versioned storage backend does not expose write cancellation".into(),
240            ));
241        }
242        Ok(())
243    }
244}
245
246/// Factory for independent sessions over one durable database.
247///
248/// A provider owns the database-level resource while each returned session
249/// owns its transaction state. This is the engine-facing extension point for
250/// `SQLite`, redb, and application-defined Key/Value stores.
251pub trait PersistentStorageProvider: Send + Sync {
252    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession>;
253
254    /// Open an autonomous session whose writes can be cancelled by the caller. Versioned factories and wrappers must preserve their catalog/backend pair and forward this construction capability.
255    fn open_session_with_cancellation(
256        &self,
257        cancellation: &uqa_core::CancellationToken,
258    ) -> StorageBackendResult<PersistentStorageSession> {
259        cancellation.check()?;
260        let session = self.open_session()?;
261        if session.backend.transaction_model().is_versioned() {
262            return Err(StorageBackendError::Other(
263                "cancellable independent storage sessions are not implemented by this provider"
264                    .into(),
265            ));
266        }
267        Ok(session)
268    }
269
270    /// 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.
271    fn open_initial_session(&self) -> StorageBackendResult<PersistentStorageSession> {
272        self.open_session()
273    }
274
275    /// 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.
276    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
277        Ok(None)
278    }
279
280    /// Protect engine-owned auxiliary files with the database's encryption
281    /// credential. Encrypted file providers must forward this capability;
282    /// `None` declares that their auxiliary storage may be unencrypted.
283    fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
284        None
285    }
286}
287
288/// Factory plus transaction surface for persistent table/index storage.
289pub trait PersistentStorageBackend: Send + Sync {
290    /// Shared allowance for this session's private state and retained query resources. Versioned wrappers must forward it; nested retained readers keep the original allowance rather than granting another limit.
291    fn retention_control(&self) -> Option<crate::read_control::StorageReadControl> {
292        None
293    }
294
295    /// Atomic auxiliary notification publication and fresh committed recovery. Session wrappers must forward this capability.
296    fn notification_publications(
297        &self,
298    ) -> Option<&dyn crate::notifications::NotificationPublicationStore> {
299        None
300    }
301
302    /// Original participant admission and logical observations, shared with the paired catalog transaction. Retained readers preserve original attribution without owning completion. Wrappers must forward this capability; SQL access-path coverage remains the execution owner's responsibility.
303    fn serializable_session(&self) -> Option<&dyn crate::mvcc::SerializableSession> {
304        None
305    }
306
307    /// Cancellation for this session's physical write admission and publication. Retained reads and rollback cleanup must remain usable after cancellation. Versioned wrappers must forward the underlying token.
308    fn write_cancellation(&self) -> Option<uqa_core::CancellationToken> {
309        None
310    }
311
312    /// Create an independent transaction context whose writes share the invoking execution's cancellation. Versioned providers must implement this without rebinding another session's token.
313    fn open_session_with_cancellation(
314        &self,
315        cancellation: &uqa_core::CancellationToken,
316    ) -> StorageBackendResult<PersistentStorageSession> {
317        cancellation.check()?;
318        if self.transaction_model().is_versioned() {
319            return Err(StorageBackendError::Other(
320                "cancellable independent storage sessions are not implemented by this backend"
321                    .into(),
322            ));
323        }
324        self.open_session()
325    }
326
327    /// Bind an independent read-only catalog/backend pair to this exact committed/private view before restoration. Both handles must retain the original snapshot and logical reader attribution without taking ownership of transaction completion. Versioned wrappers must forward this capability.
328    fn open_retained_read_session(
329        &self,
330        cancellation: &uqa_core::CancellationToken,
331    ) -> StorageBackendResult<PersistentStorageSession> {
332        cancellation.check()?;
333        Err(StorageBackendError::Other(
334            "retained read sessions are not implemented by this storage backend".into(),
335        ))
336    }
337
338    /// Transaction ownership shared with the paired catalog. Legacy providers retain serialized Engine writer admission; versioned providers keep private writes and support command refresh without ending the transaction.
339    fn transaction_model(&self) -> StorageTransactionModel {
340        StorageTransactionModel::ProviderSerialized
341    }
342
343    /// Durable identifier allocation independent of logical transaction undo. A missing capability retains serialized allocation; it cannot establish support for concurrent writers. Wrappers must forward the underlying capability.
344    fn identifier_allocator(&self) -> Option<&dyn crate::mvcc::IdentifierAllocator> {
345        None
346    }
347
348    /// Identity shared with the paired catalog's transaction context. Wrappers must delegate this when their underlying backend reports an identity.
349    fn transaction_affinity(&self) -> Option<StorageSessionAffinity> {
350        None
351    }
352
353    /// Return the stable database identity for independently constructed engines over this backend. File identities also enable cross-process row-lock coordination.
354    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
355        Ok(None)
356    }
357
358    /// Return the encryption credential for database-owned auxiliary files,
359    /// including engines constructed directly from this backend.
360    fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
361        None
362    }
363
364    /// 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.
365    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
366        Err(StorageBackendError::Other(
367            "independent sessions are not implemented for this persistent backend".into(),
368        ))
369    }
370
371    /// 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.
372    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
373        false
374    }
375
376    fn document_store(&self, table: &str) -> Box<dyn DocumentStore>;
377
378    /// Upgrade backend-owned document records before table handles are restored. Implementations must make the rewrite atomic and idempotent.
379    fn migrate_document_storage(&self) -> StorageBackendResult<()> {
380        Ok(())
381    }
382
383    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex>;
384
385    /// Upgrade backend-owned inverted-index values before table handles are
386    /// restored. Implementations must make the rewrite atomic and idempotent.
387    fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
388        Ok(())
389    }
390
391    fn vector_index(
392        &self,
393        table: &str,
394        field: &str,
395        dimensions: u32,
396        spec: VectorIndexSpec,
397        mode: VectorIndexOpenMode,
398    ) -> StorageBackendResult<Box<dyn VectorIndex>>;
399
400    fn drop_vector_index_metadata(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
401        Ok(())
402    }
403
404    /// Whether this backend maps logical `btree` indexes to durable postings.
405    fn persists_btree_indexes(&self) -> bool {
406        false
407    }
408
409    /// `Some(entries)` is a complete persisted index; `None` means it has not
410    /// been built yet and the engine must backfill it from documents once.
411    fn load_btree_index(
412        &self,
413        _table: &str,
414        _field: &crate::ValueIndexKey,
415    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
416        Ok(None)
417    }
418
419    fn btree_index_fields(&self, _table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
420        Ok(Vec::new())
421    }
422
423    /// Read one previously evaluated key on the caller's retained transaction boundary, without loading the posting collection. An unbuilt index is distinct from an absent row and a stored NULL.
424    fn read_btree_index_entry(
425        &self,
426        _table: &str,
427        _field: &crate::ValueIndexKey,
428        _doc_id: DocId,
429    ) -> StorageBackendResult<crate::ValueIndexEntry> {
430        if self.persists_btree_indexes() {
431            return Err(crate::StorageBackendError::Other(
432                "point reads of stored B-tree entries are not implemented for this backend".into(),
433            ));
434        }
435        Ok(crate::ValueIndexEntry::Unbuilt)
436    }
437
438    /// Fields whose persisted posting support was found inconsistent during a
439    /// schema migration. The engine repairs these at its explicit open-time
440    /// write boundary and clears each durable retry marker only after success.
441    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, crate::ValueIndexKey)>> {
442        Ok(Vec::new())
443    }
444
445    fn clear_btree_index_repair(
446        &self,
447        _table: &str,
448        _field: &crate::ValueIndexKey,
449    ) -> StorageBackendResult<()> {
450        Ok(())
451    }
452
453    fn replace_btree_index(
454        &self,
455        _table: &str,
456        _field: &crate::ValueIndexKey,
457        _values: &[(DocId, Value)],
458    ) -> StorageBackendResult<()> {
459        Ok(())
460    }
461
462    /// Repair sparse support differences without requiring capable backends
463    /// to rewrite every already-valid posting. The complete replacement is
464    /// supplied for the storage-neutral fallback.
465    fn repair_btree_index(
466        &self,
467        table: &str,
468        field: &crate::ValueIndexKey,
469        complete: &[(DocId, Value)],
470        _stale_doc_ids: &[DocId],
471        _missing: &[(DocId, Value)],
472    ) -> StorageBackendResult<()> {
473        self.replace_btree_index(table, field, complete)
474    }
475
476    /// Replace several complete indexes for one table atomically. Backends
477    /// may override this to share one transaction and prepared statements.
478    fn replace_btree_indexes(
479        &self,
480        table: &str,
481        indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
482    ) -> StorageBackendResult<()> {
483        for (field, values) in indexes {
484            self.replace_btree_index(table, field, values)?;
485        }
486        Ok(())
487    }
488
489    fn apply_btree_index_write(
490        &self,
491        _table: &str,
492        _doc_id: DocId,
493        _values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
494    ) -> StorageBackendResult<()> {
495        Ok(())
496    }
497
498    fn drop_btree_index(
499        &self,
500        _table: &str,
501        _field: &crate::ValueIndexKey,
502    ) -> StorageBackendResult<()> {
503        Ok(())
504    }
505
506    fn clear_btree_indexes(&self, _table: &str) -> StorageBackendResult<()> {
507        Ok(())
508    }
509
510    /// 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.
511    fn vacuum(&self) -> StorageBackendResult<()> {
512        Ok(())
513    }
514
515    fn begin_transaction(&self) -> StorageBackendResult<()>;
516
517    /// Begin a transaction whose first operation is expected to be a read.
518    /// Backends with distinct lock modes may defer write-lock acquisition;
519    /// the default preserves existing transaction semantics.
520    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
521        self.begin_transaction()
522    }
523
524    /// 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.
525    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
526        self.begin_transaction()
527    }
528
529    /// Advance command visibility without ending the active logical transaction or replaying its writes. SQL isolation controls when this is permitted. Providers without logical snapshot refresh reject the request.
530    fn refresh_transaction_snapshot(
531        &self,
532        _cancellation: &uqa_core::CancellationToken,
533    ) -> StorageBackendResult<()> {
534        Err(StorageBackendError::Other(
535            "transaction snapshot refresh is not supported by this backend".into(),
536        ))
537    }
538
539    /// Whether this session currently owns a pinned storage transaction.
540    fn in_transaction(&self) -> bool;
541
542    /// Whether the current transaction changed user data, catalog or index state. Engine uses this to enforce read-only statement transactions, including unclassified helper writes. Typed auxiliary notification publication is excluded and never grants ordinary write access.
543    fn transaction_has_written(&self) -> StorageBackendResult<bool>;
544
545    /// Backend commit generation visible to this session, when available.
546    /// A changing value invalidates session-local catalog and index caches.
547    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
548        Ok(None)
549    }
550
551    /// 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.
552    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
553        Ok(true)
554    }
555
556    /// Pin the transaction's read snapshot before cache restoration.
557    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
558        Ok(())
559    }
560
561    fn commit_transaction(&self) -> StorageBackendResult<()>;
562
563    fn rollback_transaction(&self) -> StorageBackendResult<()>;
564
565    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
566
567    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
568
569    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
570}
571
572#[cfg(all(test, unix))]
573mod identity_tests {
574    use super::*;
575
576    #[test]
577    fn dangling_database_symlink_keeps_the_target_identity_after_creation() {
578        use std::os::unix::fs::symlink;
579
580        let directory = tempfile::tempdir().unwrap();
581        let target = directory.path().join("target.db");
582        let link = directory.path().join("database.db");
583        symlink("target.db", &link).unwrap();
584
585        let before = PersistentStorageIdentity::for_database_path(&link).unwrap();
586        std::fs::File::create(&target).unwrap();
587        let after = PersistentStorageIdentity::for_database_path(&link).unwrap();
588
589        assert_eq!(before, after);
590        assert_eq!(
591            after,
592            PersistentStorageIdentity::File(target.canonicalize().unwrap())
593        );
594    }
595}
uqa_storage/backend.rs

uqa_storage/
backend.rs
