uqa_storage_sqlite/
connection.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Pooled `SQLite` connections with explicit session and transaction affinity.
8//!
9//! A [`ManagedConnection`] is a logical session. Clones share that session so
10//! catalog, document, inverted-index, and vector stores participate in the
11//! same explicit transaction. [`ManagedConnection::new_session`] creates an
12//! isolated session over the same physical connection pool. Outside explicit
13//! transactions operations check out independent connections, allowing WAL
14//! readers to make real concurrent progress.
15
16use std::path::{Path, PathBuf};
17use std::sync::Arc;
18
19use parking_lot::{Condvar, Mutex, RwLock};
20use rusqlite::{Connection, OpenFlags};
21
22use crate::compressed_vfs::{self, SQLiteCompressedContainerAnchor, SQLiteCompressionOptions};
23
24#[derive(Debug, thiserror::Error)]
25pub enum SQLiteError {
26    #[error(transparent)]
27    Memory(#[from] uqa_core::memory::MemoryError),
28    #[error(transparent)]
29    Cancelled(#[from] uqa_core::QueryCancelled),
30    #[error("text analysis failed: {0}")]
31    Analysis(#[from] uqa_analysis::AnalysisError),
32    #[error("sqlite error: {0}")]
33    SQLite(#[from] rusqlite::Error),
34    #[error("encryption key must not be empty")]
35    EmptyEncryptionKey,
36    #[error("database requires an encryption key")]
37    EncryptionKeyRequired,
38    #[error("database is not encrypted but an encryption key was provided")]
39    NotEncrypted,
40    #[error("compressed sqlite container error: {0}")]
41    CompressedContainer(String),
42    #[error("io error: {0}")]
43    Io(#[from] std::io::Error),
44    #[error("catalog migration {version} failed: {source}")]
45    Migration {
46        version: u32,
47        #[source]
48        source: rusqlite::Error,
49    },
50    #[error("invalid persisted catalog schema version `{0}`")]
51    InvalidSchemaVersion(String),
52    #[error("catalog schema version {found} is newer than this engine supports ({supported})")]
53    UnsupportedSchemaVersion { found: u32, supported: u32 },
54    #[error("corrupt document blob for `{table}` doc {doc_id} field `{field}`: {reason}")]
55    CorruptDocumentBlob {
56        table: String,
57        doc_id: u64,
58        field: String,
59        reason: String,
60    },
61    #[error("payload serialization failed: {0}")]
62    Serde(#[from] serde_json::Error),
63    #[error("storage backend error: {0}")]
64    StorageBackend(String),
65    #[error("transaction already active for this sqlite session")]
66    TransactionAlreadyActive,
67    #[error("no active transaction for this sqlite session")]
68    NoActiveTransaction,
69    #[error("sqlite transaction was aborted by an earlier storage error: {0}")]
70    TransactionAborted(String),
71    #[error("sqlite session cleanup failed: {0}")]
72    SessionCleanupFailed(String),
73    #[error("sqlite connection-pool checkout lost its connection")]
74    MissingCheckedOutConnection,
75}
76
77pub type Result<T> = std::result::Result<T, SQLiteError>;
78
79const MIN_POOL_CONNECTIONS: usize = 4;
80const MAX_POOL_CONNECTIONS: usize = 32;
81
82#[derive(Clone)]
83enum ConnectionSpec {
84    File {
85        path: PathBuf,
86        key: Option<Arc<str>>,
87    },
88    Compressed {
89        path: PathBuf,
90        compression: SQLiteCompressionOptions,
91    },
92    Memory,
93}
94
95impl ConnectionSpec {
96    fn open(&self, initialize_database: bool) -> Result<Connection> {
97        match self {
98            Self::File { path, key } => {
99                let conn = Connection::open(path)?;
100                if let Some(key) = key {
101                    ManagedConnection::apply_encryption_key(&conn, key)?;
102                }
103                if initialize_database {
104                    ManagedConnection::enable_wal(&conn)?;
105                }
106                ManagedConnection::configure_wal_connection(&conn)?;
107                Ok(conn)
108            }
109            Self::Compressed { path, compression } => {
110                let conn = Connection::open_with_flags_and_vfs(
111                    path,
112                    OpenFlags::default(),
113                    compressed_vfs::VFS_NAME,
114                )?;
115                if initialize_database {
116                    conn.pragma_update(None, "page_size", compression.page_size)?;
117                    ManagedConnection::enable_compressed_journal(&conn)?;
118                }
119                ManagedConnection::configure_compressed_connection(&conn)?;
120                Ok(conn)
121            }
122            Self::Memory => {
123                let conn = Connection::open_in_memory()?;
124                if initialize_database {
125                    ManagedConnection::enable_wal(&conn)?;
126                }
127                ManagedConnection::configure_wal_connection(&conn)?;
128                Ok(conn)
129            }
130        }
131    }
132}
133
134struct PoolState {
135    idle: Vec<Connection>,
136    open: usize,
137}
138
139struct ConnectionPool {
140    spec: ConnectionSpec,
141    max_connections: usize,
142    state: Mutex<PoolState>,
143    available: Condvar,
144    /// Stable, never-mutating connection used for `PRAGMA data_version`.
145    /// Every logical session over this pool must compare versions on this
146    /// same connection, and encrypted databases must not repeat key
147    /// derivation merely to create a request-local change monitor.
148    data_version_monitor: Mutex<Option<Connection>>,
149}
150
151impl ConnectionPool {
152    fn new(spec: ConnectionSpec, initial: Connection, max_connections: usize) -> Arc<Self> {
153        Arc::new(Self {
154            spec,
155            max_connections: max_connections.max(1),
156            state: Mutex::new(PoolState {
157                idle: vec![initial],
158                open: 1,
159            }),
160            available: Condvar::new(),
161            data_version_monitor: Mutex::new(None),
162        })
163    }
164
165    fn checkout(self: &Arc<Self>) -> Result<PooledConnection> {
166        loop {
167            let mut state = self.state.lock();
168            if let Some(connection) = state.idle.pop() {
169                return Ok(PooledConnection {
170                    pool: Arc::clone(self),
171                    connection: Some(connection),
172                });
173            }
174            if state.open < self.max_connections {
175                state.open += 1;
176                drop(state);
177                return match self.spec.open(false) {
178                    Ok(connection) => Ok(PooledConnection {
179                        pool: Arc::clone(self),
180                        connection: Some(connection),
181                    }),
182                    Err(error) => {
183                        let mut state = self.state.lock();
184                        state.open -= 1;
185                        self.available.notify_one();
186                        Err(error)
187                    }
188                };
189            }
190            self.available.wait(&mut state);
191        }
192    }
193
194    fn checkin(&self, connection: Connection) {
195        self.state.lock().idle.push(connection);
196        self.available.notify_one();
197    }
198
199    fn discard(&self) {
200        let mut state = self.state.lock();
201        state.open -= 1;
202        self.available.notify_one();
203    }
204}
205
206struct PooledConnection {
207    pool: Arc<ConnectionPool>,
208    connection: Option<Connection>,
209}
210
211impl PooledConnection {
212    fn connection(&self) -> Result<&Connection> {
213        self.connection
214            .as_ref()
215            .ok_or(SQLiteError::MissingCheckedOutConnection)
216    }
217
218    fn connection_mut(&mut self) -> Result<&mut Connection> {
219        self.connection
220            .as_mut()
221            .ok_or(SQLiteError::MissingCheckedOutConnection)
222    }
223}
224
225impl Drop for PooledConnection {
226    fn drop(&mut self) {
227        let Some(connection) = self.connection.take() else {
228            return;
229        };
230        let reusable = connection.is_autocommit() || connection.execute_batch("ROLLBACK").is_ok();
231        if reusable {
232            self.pool.checkin(connection);
233        } else {
234            self.pool.discard();
235        }
236    }
237}
238
239struct SessionState {
240    /// Read guards cover ordinary operations. Transaction lifecycle calls take
241    /// the write guard, making BEGIN/COMMIT/ROLLBACK linearizable with respect
242    /// to every operation issued through the same logical session.
243    gate: RwLock<()>,
244    transaction: Mutex<Option<PooledConnection>>,
245    transaction_failure: Mutex<Option<String>>,
246    cleanup_failure: Mutex<Option<String>>,
247}
248
249impl SessionState {
250    fn new() -> Self {
251        Self {
252            gate: RwLock::new(()),
253            transaction: Mutex::new(None),
254            transaction_failure: Mutex::new(None),
255            cleanup_failure: Mutex::new(None),
256        }
257    }
258}
259
260impl Drop for SessionState {
261    fn drop(&mut self) {
262        // `PooledConnection::drop` performs the rollback and discards the
263        // physical connection when rollback itself fails. Taking the pinned
264        // handle here therefore cannot return a broken connection to the pool.
265        self.transaction.get_mut().take();
266    }
267}
268
269/// Logical `SQLite` session backed by a bounded physical connection pool.
270/// Cloning preserves session/transaction affinity; call [`Self::new_session`]
271/// for an independently isolated transaction context.
272#[derive(Clone)]
273pub struct ManagedConnection {
274    pool: Arc<ConnectionPool>,
275    session: Arc<SessionState>,
276}
277
278impl ManagedConnection {
279    fn surface_cleanup_failure(&self) -> Result<()> {
280        if let Some(error) = self.session.cleanup_failure.lock().take() {
281            return Err(SQLiteError::SessionCleanupFailed(error));
282        }
283        Ok(())
284    }
285
286    pub fn open(path: &Path) -> Result<Self> {
287        if path == Path::new(":memory:") {
288            return Self::open_in_memory();
289        }
290        Self::open_with_optional_key(path, None)
291    }
292
293    /// Return the backing database path for a file-backed connection.
294    #[must_use]
295    pub fn database_path(&self) -> Option<&Path> {
296        match &self.pool.spec {
297            ConnectionSpec::File { path, .. } | ConnectionSpec::Compressed { path, .. } => {
298                Some(path)
299            }
300            ConnectionSpec::Memory => None,
301        }
302    }
303
304    pub fn open_encrypted(path: &Path, key: &str) -> Result<Self> {
305        Self::open_with_optional_key(path, Some(key))
306    }
307
308    pub fn open_compressed(path: &Path, compression: SQLiteCompressionOptions) -> Result<Self> {
309        Self::open_compressed_with_optional_key(path, compression, None, None)
310    }
311
312    pub fn open_compressed_encrypted(
313        path: &Path,
314        key: &str,
315        compression: SQLiteCompressionOptions,
316    ) -> Result<Self> {
317        if key.is_empty() {
318            return Err(SQLiteError::EmptyEncryptionKey);
319        }
320        Self::open_compressed_with_optional_key(path, compression, Some(key), None)
321    }
322
323    /// Open an encrypted compressed database while enforcing an exact trusted
324    /// external state anchor in the VFS main-file open path.
325    pub fn open_compressed_encrypted_with_anchor(
326        path: &Path,
327        key: &str,
328        compression: SQLiteCompressionOptions,
329        trusted_anchor: SQLiteCompressedContainerAnchor,
330    ) -> Result<Self> {
331        if key.is_empty() {
332            return Err(SQLiteError::EmptyEncryptionKey);
333        }
334        Self::open_compressed_with_optional_key(path, compression, Some(key), Some(trusted_anchor))
335    }
336
337    fn open_with_optional_key(path: &Path, key: Option<&str>) -> Result<Self> {
338        let spec = ConnectionSpec::File {
339            path: path.to_path_buf(),
340            key: key.map(Arc::from),
341        };
342        Self::from_spec(spec, default_pool_connections())
343    }
344
345    fn open_compressed_with_optional_key(
346        path: &Path,
347        compression: SQLiteCompressionOptions,
348        key: Option<&str>,
349        trusted_anchor: Option<SQLiteCompressedContainerAnchor>,
350    ) -> Result<Self> {
351        let compression = compression
352            .validate()
353            .map_err(SQLiteError::CompressedContainer)?;
354        match trusted_anchor {
355            Some(anchor) => compressed_vfs::register_database_with_anchor(
356                path,
357                compression,
358                key.ok_or(SQLiteError::EncryptionKeyRequired)?,
359                anchor,
360            ),
361            None => compressed_vfs::register_database(path, compression, key),
362        }
363        .map_err(SQLiteError::CompressedContainer)?;
364        let spec = ConnectionSpec::Compressed {
365            path: path.to_path_buf(),
366            compression,
367        };
368        Self::from_spec(spec, default_pool_connections())
369    }
370
371    pub fn open_in_memory() -> Result<Self> {
372        // Independent `:memory:` connections do not share a database. Keep a
373        // one-connection pool for this special target; file-backed databases
374        // use the real multi-connection pool.
375        Self::from_spec(ConnectionSpec::Memory, 1)
376    }
377
378    fn from_spec(spec: ConnectionSpec, max_connections: usize) -> Result<Self> {
379        let initial = spec.open(true)?;
380        Ok(Self {
381            pool: ConnectionPool::new(spec, initial, max_connections),
382            session: Arc::new(SessionState::new()),
383        })
384    }
385
386    fn apply_encryption_key(conn: &Connection, key: &str) -> Result<()> {
387        if key.is_empty() {
388            return Err(SQLiteError::EmptyEncryptionKey);
389        }
390        conn.pragma_update(None, "key", key)?;
391        Ok(())
392    }
393
394    fn enable_wal(conn: &Connection) -> Result<()> {
395        conn.pragma_update(None, "journal_mode", "WAL")?;
396        Ok(())
397    }
398
399    fn configure_wal_connection(conn: &Connection) -> Result<()> {
400        // 5s busy timeout absorbs short contention without surfacing
401        // SQLITE_BUSY; long contention is a real bug.
402        conn.busy_timeout(std::time::Duration::from_secs(5))?;
403        // Synchronous=NORMAL is the recommended pairing with WAL: safe
404        // against power loss, faster than FULL.
405        conn.pragma_update(None, "synchronous", "NORMAL")?;
406        // Foreign keys are off by default; turn on so per-table cleanup
407        // can use ON DELETE CASCADE later.
408        conn.pragma_update(None, "foreign_keys", "ON")?;
409        Ok(())
410    }
411
412    fn enable_compressed_journal(conn: &Connection) -> Result<()> {
413        // The compressed VFS implements the byte-addressed database file.
414        // Rollback journals stay raw because they are short-lived commit
415        // machinery; compressing them only adds autocommit write amplification.
416        // WAL requires shared-memory VFS methods, so compressed databases use
417        // SQLite's rollback journal and keep temp storage in memory.
418        conn.pragma_update(None, "journal_mode", "DELETE")?;
419        Ok(())
420    }
421
422    fn configure_compressed_connection(conn: &Connection) -> Result<()> {
423        conn.busy_timeout(std::time::Duration::from_secs(5))?;
424        conn.pragma_update(None, "synchronous", "FULL")?;
425        conn.pragma_update(None, "temp_store", "MEMORY")?;
426        conn.pragma_update(None, "foreign_keys", "ON")?;
427        Ok(())
428    }
429
430    /// Create an independent logical session over the same database pool.
431    /// Explicit transactions started on either session are isolated and never
432    /// capture operations issued through the other session.
433    #[must_use]
434    pub fn new_session(&self) -> Self {
435        Self {
436            pool: Arc::clone(&self.pool),
437            session: Arc::new(SessionState::new()),
438        }
439    }
440
441    /// Whether this logical session currently owns a pinned transaction
442    /// connection.
443    #[must_use]
444    pub fn in_transaction(&self) -> bool {
445        let _gate = self.session.gate.read();
446        self.session.transaction.lock().is_some()
447    }
448
449    /// Whether the currently pinned transaction has upgraded to a write
450    /// transaction. Callers use this to enforce read-only execution
451    /// boundaries before COMMIT rather than inferring writes from SQL shape.
452    pub fn transaction_has_written(&self) -> Result<bool> {
453        let _gate = self.session.gate.read();
454        let transaction = self.session.transaction.lock();
455        let transaction = transaction
456            .as_ref()
457            .ok_or(SQLiteError::NoActiveTransaction)?;
458        Ok(matches!(
459            transaction.connection()?.transaction_state(Some("main"))?,
460            rusqlite::TransactionState::Write
461        ))
462    }
463
464    /// Whether this session's independent [`Self::data_version`] monitor can
465    /// read without contending with the currently pinned transaction.
466    ///
467    /// A rollback-journal writer's pending lock blocks new readers while waiting for existing readers to finish. A compressed read transaction must therefore also avoid the independent monitor: its own shared lock may be preventing that waiting writer from proceeding. Callers refresh through the pinned connection instead. WAL sessions and compressed sessions without a pinned transaction permit the independent monitor.
468    pub fn data_version_monitor_is_nonblocking(&self) -> Result<bool> {
469        if !matches!(&self.pool.spec, ConnectionSpec::Compressed { .. }) {
470            return Ok(true);
471        }
472        let _gate = self.session.gate.read();
473        Ok(self.session.transaction.lock().is_none())
474    }
475
476    /// Whether one pooled connection may retain a read snapshot while another writes. Plain and encrypted `SQLite` databases use WAL; compressed containers use rollback journaling and therefore require a detached engine snapshot before writer promotion.
477    #[must_use]
478    pub fn supports_concurrent_pinned_read_and_write(&self) -> bool {
479        !matches!(&self.pool.spec, ConnectionSpec::Compressed { .. })
480    }
481
482    /// Database change counter observed on one stable connection shared by
483    /// every logical session over this pool. The value changes when another
484    /// `SQLite` connection commits. In-memory databases have no independent
485    /// connections and therefore return `None`.
486    pub fn data_version(&self) -> Result<Option<u64>> {
487        if matches!(&self.pool.spec, ConnectionSpec::Memory) {
488            return Ok(None);
489        }
490        let mut monitor = self.pool.data_version_monitor.lock();
491        if monitor.is_none() {
492            *monitor = Some(self.pool.spec.open(false)?);
493        }
494        let monitor = monitor.as_ref().ok_or_else(|| {
495            SQLiteError::StorageBackend(
496                "data-version monitor was not initialized after opening it".into(),
497            )
498        })?;
499        let version: i64 = monitor.pragma_query_value(None, "data_version", |row| row.get(0))?;
500        let version = u64::try_from(version).map_err(|_| {
501            SQLiteError::StorageBackend(format!(
502                "SQLite returned a negative PRAGMA data_version: {version}"
503            ))
504        })?;
505        Ok(Some(version))
506    }
507
508    /// Establish the database snapshot for the active transaction without
509    /// depending on the caller's first user query. `BEGIN DEFERRED` alone does
510    /// not start a read transaction, so a writer could otherwise commit after
511    /// the engine checks its cache generations but before the first catalog
512    /// read. Reading `sqlite_schema` is database-wide and keeps the operation
513    /// independent of any application table.
514    pub fn pin_transaction_snapshot(&self) -> Result<()> {
515        self.with(|connection| {
516            if connection.is_autocommit() {
517                return Err(SQLiteError::NoActiveTransaction);
518            }
519            let _: i64 =
520                connection.query_row("SELECT COUNT(*) FROM sqlite_schema", [], |row| row.get(0))?;
521            Ok(())
522        })
523    }
524
525    /// Run a closure using this session. Outside a transaction the closure
526    /// checks out a pooled connection; inside a transaction every clone is
527    /// routed to the session's pinned connection.
528    pub fn with<R>(&self, f: impl FnOnce(&Connection) -> Result<R>) -> Result<R> {
529        self.surface_cleanup_failure()?;
530        let _gate = self.session.gate.read();
531        let transaction = self.session.transaction.lock();
532        if let Some(connection) = transaction.as_ref() {
533            if let Some(error) = self.session.transaction_failure.lock().as_ref() {
534                return Err(SQLiteError::TransactionAborted(error.clone()));
535            }
536            let result = f(connection.connection()?);
537            if let Err(error) = &result {
538                let mut failure = self.session.transaction_failure.lock();
539                if failure.is_none() {
540                    *failure = Some(error.to_string());
541                }
542            }
543            return result;
544        }
545        drop(transaction);
546        let connection = self.pool.checkout()?;
547        f(connection.connection()?)
548    }
549
550    pub fn with_mut<R>(&self, f: impl FnOnce(&mut Connection) -> Result<R>) -> Result<R> {
551        self.surface_cleanup_failure()?;
552        let _gate = self.session.gate.read();
553        let mut transaction = self.session.transaction.lock();
554        if let Some(connection) = transaction.as_mut() {
555            if let Some(error) = self.session.transaction_failure.lock().as_ref() {
556                return Err(SQLiteError::TransactionAborted(error.clone()));
557            }
558            let result = f(connection.connection_mut()?);
559            if let Err(error) = &result {
560                let mut failure = self.session.transaction_failure.lock();
561                if failure.is_none() {
562                    *failure = Some(error.to_string());
563                }
564            }
565            return result;
566        }
567        drop(transaction);
568        let mut connection = self.pool.checkout()?;
569        f(connection.connection_mut()?)
570    }
571
572    /// Rewrite the `SQLite` database into its minimum-sized file. `SQLite` requires `VACUUM` to run in autocommit mode, so the session write gate makes the transaction check and maintenance command one atomic session operation.
573    pub fn vacuum(&self) -> Result<()> {
574        self.surface_cleanup_failure()?;
575        let _gate = self.session.gate.write();
576        if self.session.transaction.lock().is_some() {
577            return Err(SQLiteError::TransactionAlreadyActive);
578        }
579        let connection = self.pool.checkout()?;
580        connection.connection()?.execute_batch("VACUUM")?;
581        Ok(())
582    }
583
584    /// Open an explicit (non-deferred) transaction. Subsequent
585    /// auto-commit hosts (catalog writes, FTS index updates, ...) all
586    /// flow through the same connection so the transaction enclosing
587    /// them is honoured. Use [`Self::commit_transaction`] /
588    /// [`Self::rollback_transaction`] / [`Self::savepoint`] etc. for
589    /// the lifecycle.
590    pub fn begin_transaction(&self) -> Result<()> {
591        self.begin_transaction_with("BEGIN IMMEDIATE")
592    }
593
594    /// Open a deferred transaction. Read-only SQL statements use this mode
595    /// so WAL readers do not take the single writer reservation; if a scalar
596    /// routine performs a write, `SQLite` upgrades the same transaction and
597    /// still preserves the statement's atomic boundary.
598    pub fn begin_deferred_transaction(&self) -> Result<()> {
599        self.begin_transaction_with("BEGIN DEFERRED")
600    }
601
602    fn begin_transaction_with(&self, statement: &str) -> Result<()> {
603        self.surface_cleanup_failure()?;
604        let _gate = self.session.gate.write();
605        let mut transaction = self.session.transaction.lock();
606        if transaction.is_some() {
607            return Err(SQLiteError::TransactionAlreadyActive);
608        }
609        let connection = self.pool.checkout()?;
610        connection.connection()?.execute_batch(statement)?;
611        self.session.transaction_failure.lock().take();
612        *transaction = Some(connection);
613        Ok(())
614    }
615
616    pub fn commit_transaction(&self) -> Result<()> {
617        self.surface_cleanup_failure()?;
618        self.finish_transaction("COMMIT")
619    }
620
621    pub fn rollback_transaction(&self) -> Result<()> {
622        self.surface_cleanup_failure()?;
623        self.finish_transaction("ROLLBACK")
624    }
625
626    /// Drop-only transaction cleanup. A rollback error is recorded on the
627    /// logical session and is returned by its next operation instead of being
628    /// mistaken for a successful rollback.
629    pub(crate) fn rollback_transaction_on_drop(&self) {
630        if let Err(error) = self.finish_transaction("ROLLBACK") {
631            let mut failure = self.session.cleanup_failure.lock();
632            if failure.is_none() {
633                *failure = Some(error.to_string());
634            }
635        }
636    }
637
638    fn finish_transaction(&self, statement: &str) -> Result<()> {
639        let _gate = self.session.gate.write();
640        let mut transaction = self.session.transaction.lock();
641        let connection = transaction
642            .as_ref()
643            .ok_or(SQLiteError::NoActiveTransaction)?;
644        if statement == "COMMIT" {
645            // Materialize the failure before entering the branch. Holding the
646            // mutex guard created by an `if let` scrutinee until the end of
647            // the branch would deadlock when cleanup takes the same lock.
648            let transaction_failure = self.session.transaction_failure.lock().clone();
649            if let Some(error) = transaction_failure {
650                if let Err(rollback_error) = connection.connection()?.execute_batch("ROLLBACK") {
651                    transaction.take();
652                    self.session.transaction_failure.lock().take();
653                    return Err(SQLiteError::SQLite(rollback_error));
654                }
655                transaction.take();
656                self.session.transaction_failure.lock().take();
657                return Err(SQLiteError::TransactionAborted(error));
658            }
659        }
660        if let Err(error) = connection.connection()?.execute_batch(statement) {
661            transaction.take();
662            self.session.transaction_failure.lock().take();
663            return Err(SQLiteError::SQLite(error));
664        }
665        transaction.take();
666        self.session.transaction_failure.lock().take();
667        Ok(())
668    }
669
670    fn with_transaction<R>(&self, f: impl FnOnce(&Connection) -> Result<R>) -> Result<R> {
671        let _gate = self.session.gate.read();
672        let transaction = self.session.transaction.lock();
673        let connection = transaction
674            .as_ref()
675            .ok_or(SQLiteError::NoActiveTransaction)?;
676        f(connection.connection()?)
677    }
678
679    pub fn savepoint(&self, name: &str) -> Result<()> {
680        self.surface_cleanup_failure()?;
681        let stmt = format!("SAVEPOINT \"{}\"", name.replace('"', "\"\""));
682        self.with_transaction(|c| {
683            c.execute_batch(&stmt)?;
684            Ok(())
685        })
686    }
687
688    pub fn release_savepoint(&self, name: &str) -> Result<()> {
689        self.surface_cleanup_failure()?;
690        let stmt = format!("RELEASE SAVEPOINT \"{}\"", name.replace('"', "\"\""));
691        self.with_transaction(|c| {
692            c.execute_batch(&stmt)?;
693            Ok(())
694        })
695    }
696
697    pub fn rollback_to_savepoint(&self, name: &str) -> Result<()> {
698        self.surface_cleanup_failure()?;
699        let stmt = format!("ROLLBACK TO SAVEPOINT \"{}\"", name.replace('"', "\"\""));
700        self.with_transaction(|c| {
701            c.execute_batch(&stmt)?;
702            Ok(())
703        })?;
704        self.session.transaction_failure.lock().take();
705        Ok(())
706    }
707}
708
709fn default_pool_connections() -> usize {
710    std::thread::available_parallelism()
711        .map_or(MIN_POOL_CONNECTIONS, |parallelism| parallelism.get() * 2)
712        .clamp(MIN_POOL_CONNECTIONS, MAX_POOL_CONNECTIONS)
713}
714
715#[cfg(test)]
716mod tests;
717
718impl From<SQLiteError> for uqa_storage::StorageBackendError {
719    fn from(source: SQLiteError) -> Self {
720        match source {
721            SQLiteError::Memory(error) => Self::Memory(error),
722            SQLiteError::Cancelled(error) => Self::Cancelled(error),
723            source => Self::backend("SQLite", source),
724        }
725    }
726}
727
728impl From<SQLiteError> for uqa_storage::TransactionError {
729    fn from(source: SQLiteError) -> Self {
730        Self::Storage(source.into())
731    }
732}
733
734impl From<uqa_storage::StorageBackendError> for SQLiteError {
735    fn from(error: uqa_storage::StorageBackendError) -> Self {
736        use uqa_storage::StorageBackendError;
737        match error {
738            StorageBackendError::Memory(error) => Self::Memory(error),
739            StorageBackendError::Cancelled(error) => Self::Cancelled(error),
740            StorageBackendError::Analysis(error) => Self::Analysis(error),
741            StorageBackendError::Serde(error) => Self::Serde(error),
742            StorageBackendError::Backend { backend, source } => match source.downcast::<Self>() {
743                Ok(error) => *error,
744                Err(source) => Self::StorageBackend(format!("{backend} storage failed: {source}")),
745            },
746            StorageBackendError::Other(message) => Self::StorageBackend(message),
747        }
748    }
749}
uqa_storage_sqlite/connection.rs

uqa_storage_sqlite/
connection.rs
