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

uqa_storage/sqlite/
connection.rs
