rightkit_db/
lib.rs

1// Unsafe is confined to the host-name FFI in `host`.
2#![deny(unsafe_code)]
3
4#[cfg(all(feature = "sqlcipher", feature = "sqlcipher-system"))]
5compile_error!("rightkit-db features `sqlcipher` and `sqlcipher-system` are mutually exclusive; enable exactly one");
6
7use rusqlite::{Connection, OpenFlags, Transaction, TransactionBehavior};
8use std::ffi::OsString;
9use std::ops::Deref;
10use std::path::{Path, PathBuf};
11use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
12use std::time::Duration;
13use zeroize::Zeroizing;
14
15mod backup;
16#[allow(unsafe_code)]
17mod host;
18pub use host::per_host_path;
19/// The exact `rusqlite` this crate links (`links = "sqlite3"` allows one per build). Apps name
20/// `Connection`, `params!`, `Row`, `OptionalExtension`, `TransactionBehavior` through this path
21/// instead of declaring their own `rusqlite` dependency.
22pub use rusqlite;
23mod migrate;
24mod recover;
25
26pub use backup::{online_backup_to, restore_from, BackupProgress};
27pub use migrate::{
28    add_column_if_missing, column_exists, migrate, migration_version, pending_marker, rollback_to,
29    rollback_to_with_storage, user_version, MigrateOptions, Migration, MigrationReport,
30    PendingMarker, Step, VersionStorage,
31};
32#[cfg(feature = "cipher")]
33pub use recover::rekey;
34pub use recover::{quarantine_database, OpenOutcome, RecoveryPolicy};
35
36const BUSY_TIMEOUT_MS: u64 = 5_000;
37const JOURNAL_SIZE_LIMIT: i64 = 67_108_864;
38
39pub type Result<T> = std::result::Result<T, DbError>;
40
41#[derive(Debug, thiserror::Error)]
42pub enum DbError {
43    #[error("SQLite error: {0}")]
44    Sqlite(#[from] rusqlite::Error),
45    #[error("a SQLCipher key requires the rightkit-db `sqlcipher` or `sqlcipher-system` feature")]
46    SqlcipherFeatureRequired,
47    #[error("SQLCipher key contains characters that are unsafe for PRAGMA interpolation")]
48    InvalidSqlcipherKey,
49    #[error("database integrity check failed: {0}")]
50    Integrity(String),
51    #[error("read-pool capacity must be greater than zero")]
52    InvalidPoolCapacity,
53    #[error("database path has no file name")]
54    InvalidDatabasePath,
55    #[error("database is at schema v{current}, newer than this build understands (v{max}) and not forward compatible")]
56    NewerDatabase { current: i64, max: i64 },
57    #[error("invalid migration list: {0}")]
58    InvalidMigrations(String),
59    #[error("migration v{version} ({name}) failed and was rolled back: {source}")]
60    MigrationFailed {
61        version: i64,
62        name: &'static str,
63        #[source]
64        source: rusqlite::Error,
65    },
66    #[error("cannot roll back: {0}")]
67    Rollback(String),
68    #[error("database backup failed: {0}")]
69    Backup(String),
70    #[error("no candidate key opened the database")]
71    NoKeyOpened,
72    #[error(transparent)]
73    Fs(#[from] rightkit_fs::FsError),
74}
75
76#[derive(Clone, Copy, Debug, Eq, PartialEq)]
77pub enum CacheProfile {
78    Default,
79    LargeWorkingSet,
80}
81
82#[derive(Clone, Copy, Debug, Eq, PartialEq)]
83pub enum DurabilityProfile {
84    Full,
85    Normal,
86}
87
88/// How the journal mode is chosen. WAL needs working shared-memory locking, which network
89/// filesystems often lack and which SQLite will silently downgrade; the default therefore
90/// *verifies* the mode that took effect and falls back to a rollback journal.
91#[derive(Clone, Copy, Debug, Eq, PartialEq)]
92#[non_exhaustive]
93pub enum JournalSelect {
94    /// WAL when the filesystem honors it and is not a detected network mount; else a
95    /// rollback journal (TRUNCATE with [`DbOptions::per_host_fallback`], else DELETE).
96    WalWithFallback,
97    /// WAL or fail (no fallback).
98    WalStrict,
99    /// Rollback journal, deleted after each transaction.
100    Delete,
101    /// Rollback journal, truncated after each transaction.
102    Truncate,
103}
104
105impl JournalSelect {
106    fn is_rollback(self) -> bool {
107        matches!(self, JournalSelect::Delete | JournalSelect::Truncate)
108    }
109}
110
111/// Environment kill-switch (`wal` | `delete` | `truncate`) for a misclassified mount in the
112/// field. An app's own variable ([`DbOptions::journal_env`]) is checked first.
113pub const JOURNAL_ENV: &str = "RIGHTKIT_SQLITE_JOURNAL_MODE";
114
115/// Post-key, pre-pragma per-connection initializer (register functions, collations, ...).
116pub type ConnInit = Arc<dyn Fn(&Connection) -> rusqlite::Result<()> + Send + Sync>;
117
118#[derive(Clone)]
119pub struct DbOptions {
120    pub path: PathBuf,
121    /// SQLite busy wait, installed before key verification or any bootstrap statement.
122    /// Default: five seconds.
123    pub busy_timeout: Duration,
124    /// SQLCipher passphrase-style key (base64-ish charset). See also [`DbOptions::raw_key`].
125    pub key: Option<Zeroizing<String>>,
126    /// SQLCipher raw 256-bit key, applied as `x'<hex>'` (no KDF). Takes precedence over `key`.
127    pub raw_key: Option<Zeroizing<[u8; 32]>>,
128    pub profile: CacheProfile,
129    pub durability: DurabilityProfile,
130    pub foreign_keys: bool,
131    pub integrity_check_on_open: bool,
132    pub journal: JournalSelect,
133    /// Narrow the database and its sidecars to owner-only (0600) on unix. Default on.
134    pub owner_only: bool,
135    pub init: Option<ConnInit>,
136    /// The app's own kill-switch variable (e.g. `CODERIGHT_SQLITE_JOURNAL_MODE`), checked
137    /// before [`JOURNAL_ENV`].
138    pub journal_env: Option<String>,
139    /// When a rollback journal is in force (network mount, kill switch, or WAL refused),
140    /// open the per-host sibling [`per_host_path`] instead of the requested file, so hosts
141    /// sharing a network directory never share one rollback-journaled database. The
142    /// requested file is not created. Default off.
143    pub per_host_fallback: bool,
144}
145
146impl DbOptions {
147    pub fn new(path: impl Into<PathBuf>) -> Self {
148        Self {
149            path: path.into(),
150            busy_timeout: Duration::from_millis(BUSY_TIMEOUT_MS),
151            key: None,
152            raw_key: None,
153            profile: CacheProfile::Default,
154            durability: DurabilityProfile::Normal,
155            foreign_keys: false,
156            integrity_check_on_open: false,
157            journal: JournalSelect::WalWithFallback,
158            owner_only: true,
159            init: None,
160            journal_env: None,
161            per_host_fallback: false,
162        }
163    }
164
165    pub fn with_journal_env(mut self, name: impl Into<String>) -> Self {
166        self.journal_env = Some(name.into());
167        self
168    }
169
170    pub fn with_per_host_fallback(mut self, on: bool) -> Self {
171        self.per_host_fallback = on;
172        self
173    }
174
175    /// The file connections actually open: the requested path, or its per-host sibling when
176    /// [`Self::per_host_fallback`] applies (a rollback journal is in force, or a writer
177    /// already fell back there).
178    pub fn effective_path(&self) -> PathBuf {
179        if !self.per_host_fallback {
180            return self.path.clone();
181        }
182        let sibling = per_host_path(&self.path);
183        if effective_journal(self).is_rollback() || (sibling.exists() && !self.path.exists()) {
184            sibling
185        } else {
186            self.path.clone()
187        }
188    }
189
190    pub fn with_passphrase(mut self, key: impl Into<String>) -> Self {
191        self.key = Some(Zeroizing::new(key.into()));
192        self.raw_key = None;
193        self
194    }
195
196    pub fn with_raw_key(mut self, key: [u8; 32]) -> Self {
197        self.raw_key = Some(Zeroizing::new(key));
198        self.key = None;
199        self
200    }
201
202    pub fn is_keyed(&self) -> bool {
203        self.key.is_some() || self.raw_key.is_some()
204    }
205
206    pub fn open_writer(&self) -> Result<Connection> {
207        self.open(ConnectionKind::Writer)
208    }
209
210    pub fn open_reader(&self) -> Result<Connection> {
211        self.open(ConnectionKind::Reader)
212    }
213
214    /// Open exactly `path` read-only, without creating it or running mutating bootstrap.
215    /// URI parameters are passed to SQLite unchanged. Journal fallback, permissions,
216    /// `init`, optimization, and migrations are skipped; key and timeout still apply.
217    pub fn open_read_only(&self) -> Result<Connection> {
218        self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_ONLY)
219    }
220
221    /// Open exactly `path` read-write without CREATE or mutating bootstrap.
222    /// Uses the same minimal setup as [`Self::open_read_only`].
223    pub fn open_existing_writer(&self) -> Result<Connection> {
224        self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_WRITE)
225    }
226
227    /// Open a private, ephemeral database. `path`, filesystem permissions, journal
228    /// selection/fallback, and mmap are ignored; other bootstrap options still apply.
229    pub fn open_in_memory(&self) -> Result<Connection> {
230        let key = self.key_pragma_value()?;
231        let conn = Connection::open_in_memory()?;
232        self.initialize(&conn, key.as_ref().map(|k| k.as_str()))?;
233        configure_connection(&conn, self, ConnectionKind::Memory, None)?;
234        if self.integrity_check_on_open {
235            integrity_check(&conn, self.is_keyed())?;
236        }
237        Ok(conn)
238    }
239
240    fn open_existing_at(&self, path: &Path, access: OpenFlags) -> Result<Connection> {
241        let key = self.key_pragma_value()?;
242        let conn = Connection::open_with_flags(
243            path,
244            access | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX,
245        )?;
246        conn.busy_timeout(self.busy_timeout)?;
247        apply_key_first(&conn, key.as_ref().map(|k| k.as_str()))?;
248        verify_key(&conn)?;
249        if self.integrity_check_on_open {
250            integrity_check(&conn, self.is_keyed())?;
251        }
252        Ok(conn)
253    }
254
255    fn initialize(&self, conn: &Connection, key: Option<&str>) -> Result<()> {
256        conn.busy_timeout(self.busy_timeout)?;
257        apply_key_first(conn, key)?;
258        verify_key(conn)?;
259        if let Some(init) = &self.init {
260            init(conn)?;
261        }
262        Ok(())
263    }
264
265    pub fn open_read_pool(&self, cap: usize) -> Result<ReadPool> {
266        ReadPool::open(self.clone(), cap)
267    }
268
269    fn open(&self, kind: ConnectionKind) -> Result<Connection> {
270        let path = self.effective_path();
271        let existed = path.exists();
272        // The per-host sibling is always rollback-journaled, whoever opens it.
273        let forced = (path != self.path).then(|| match effective_journal(self) {
274            mode if mode.is_rollback() => mode,
275            _ => fallback_mode(self),
276        });
277        let (conn, journal) = self.open_at(&path, kind, forced)?;
278        // WAL refused at runtime: with per-host fallback, the rollback journal belongs on the
279        // per-host sibling, and a file this open created at the requested path goes away.
280        if journal == Opened::FellBack && self.per_host_fallback && path == self.path {
281            drop(conn);
282            if !existed {
283                for suffix in ["", "-journal", "-wal", "-shm"] {
284                    let mut name = OsString::from(path.as_os_str());
285                    name.push(suffix);
286                    let _ = std::fs::remove_file(PathBuf::from(name));
287                }
288            }
289            let sibling = per_host_path(&self.path);
290            return Ok(self
291                .open_at(&sibling, kind, Some(JournalSelect::Truncate))?
292                .0);
293        }
294        Ok(conn)
295    }
296
297    fn open_at(
298        &self,
299        path: &Path,
300        kind: ConnectionKind,
301        forced: Option<JournalSelect>,
302    ) -> Result<(Connection, Opened)> {
303        let key = self.key_pragma_value()?;
304        let flags = OpenFlags::SQLITE_OPEN_READ_WRITE
305            | OpenFlags::SQLITE_OPEN_CREATE
306            | OpenFlags::SQLITE_OPEN_NO_MUTEX;
307        let conn = Connection::open_with_flags(path, flags)?;
308        // Narrow the main file before any write so WAL/SHM sidecars inherit owner-only.
309        if self.owner_only {
310            rightkit_fs::restrict_owner_only(path)?;
311        }
312        self.initialize(&conn, key.as_ref().map(|k| k.as_str()))?;
313        let journal = configure_connection(&conn, self, kind, forced)?;
314        if self.integrity_check_on_open {
315            integrity_check(&conn, self.is_keyed())?;
316        }
317        if self.owner_only {
318            harden_sidecars(path)?;
319        }
320        Ok((conn, journal))
321    }
322
323    /// The `PRAGMA key` right-hand side (already quoted), validated against injection.
324    fn key_pragma_value(&self) -> Result<Option<Zeroizing<String>>> {
325        if let Some(raw) = &self.raw_key {
326            #[cfg(not(feature = "cipher"))]
327            {
328                let _ = raw;
329                return Err(DbError::SqlcipherFeatureRequired);
330            }
331            #[cfg(feature = "cipher")]
332            {
333                let hex: String = raw.iter().map(|b| format!("{b:02x}")).collect();
334                return Ok(Some(Zeroizing::new(format!("\"x'{hex}'\""))));
335            }
336        }
337        let key = self.key.as_ref().map(|key| key.as_str());
338        validate_key(key)?;
339        Ok(key.map(|k| Zeroizing::new(format!("'{k}'"))))
340    }
341}
342
343/// Default-options read-only open; never creates a missing database.
344pub fn open_read_only(path: impl AsRef<Path>) -> Result<Connection> {
345    DbOptions::new(path.as_ref()).open_read_only()
346}
347
348/// Default-options read-write open; never creates a missing database.
349pub fn open_existing_writer(path: impl AsRef<Path>) -> Result<Connection> {
350    DbOptions::new(path.as_ref()).open_existing_writer()
351}
352
353/// Default-options private in-memory database.
354pub fn open_in_memory() -> Result<Connection> {
355    DbOptions::new(":memory:").open_in_memory()
356}
357
358fn harden_sidecars(db: &Path) -> Result<()> {
359    for suffix in ["-wal", "-shm"] {
360        let mut name = OsString::from(db.file_name().ok_or(DbError::InvalidDatabasePath)?);
361        name.push(suffix);
362        rightkit_fs::restrict_owner_only(&db.with_file_name(name))?;
363    }
364    Ok(())
365}
366
367#[derive(Clone, Copy)]
368enum ConnectionKind {
369    Writer,
370    Reader,
371    Memory,
372}
373
374/// What journal setup did.
375#[derive(Clone, Copy, Debug, Eq, PartialEq)]
376enum Opened {
377    AsRequested,
378    /// WAL was wanted but not honored; a rollback journal is in force.
379    FellBack,
380}
381
382fn configure_connection(
383    conn: &Connection,
384    options: &DbOptions,
385    kind: ConnectionKind,
386    forced: Option<JournalSelect>,
387) -> Result<Opened> {
388    let opened = if matches!(kind, ConnectionKind::Memory) {
389        Opened::AsRequested
390    } else {
391        configure_journal(conn, options, kind, forced)?
392    };
393    let synchronous = match options.durability {
394        DurabilityProfile::Full => "FULL",
395        DurabilityProfile::Normal => "NORMAL",
396    };
397    conn.pragma_update(None, "synchronous", synchronous)?;
398    conn.pragma_update(None, "journal_size_limit", JOURNAL_SIZE_LIMIT)?;
399    conn.pragma_update(None, "temp_store", "MEMORY")?;
400    if options.foreign_keys {
401        conn.pragma_update(None, "foreign_keys", true)?;
402    }
403    if options.profile == CacheProfile::LargeWorkingSet {
404        conn.pragma_update(None, "cache_size", -65_536_i64)?;
405        if !matches!(kind, ConnectionKind::Memory) {
406            conn.pragma_update(None, "mmap_size", 268_435_456_i64)?;
407        }
408    }
409    optimize_on_open(conn)?;
410    if matches!(kind, ConnectionKind::Reader) {
411        conn.pragma_update(None, "query_only", true)?;
412    }
413    Ok(opened)
414}
415
416fn set_rollback(conn: &Connection, mode: JournalSelect) -> Result<()> {
417    let pragma = if mode == JournalSelect::Truncate {
418        "PRAGMA journal_mode=TRUNCATE"
419    } else {
420        "PRAGMA journal_mode=DELETE"
421    };
422    conn.query_row(pragma, [], |row| row.get::<_, String>(0))?;
423    Ok(())
424}
425
426fn configure_journal(
427    conn: &Connection,
428    options: &DbOptions,
429    kind: ConnectionKind,
430    forced: Option<JournalSelect>,
431) -> Result<Opened> {
432    let wanted = forced.unwrap_or_else(|| effective_journal(options));
433    if wanted.is_rollback() {
434        // Rollback modes are per connection: a reader left in the default DELETE mode would
435        // try to delete a TRUNCATE writer's open journal (IOERR_DELETE on Windows).
436        set_rollback(conn, wanted)?;
437        return Ok(Opened::AsRequested);
438    }
439    let honored = conn
440        .query_row("PRAGMA journal_mode=WAL", [], |row| row.get::<_, String>(0))
441        .map(|mode| mode.eq_ignore_ascii_case("wal"))
442        .unwrap_or(false)
443        // `journal_mode=WAL` can report success before the -shm/-wal sidecars are usable;
444        // one empty write transaction proves the sidecars work before anything depends on them.
445        && (matches!(kind, ConnectionKind::Reader)
446            || conn.execute_batch("BEGIN IMMEDIATE; ROLLBACK;").is_ok());
447    if honored {
448        return Ok(Opened::AsRequested);
449    }
450    if wanted == JournalSelect::WalStrict {
451        return Err(DbError::Sqlite(rusqlite::Error::SqliteFailure(
452            rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR),
453            Some("WAL journal mode was not honored by this filesystem".into()),
454        )));
455    }
456    let _ = conn.execute_batch("ROLLBACK;");
457    set_rollback(conn, fallback_mode(options))?;
458    Ok(Opened::FellBack)
459}
460
461/// The rollback journal used when WAL is unavailable: TRUNCATE with per-host fallback
462/// (CodeRight's network mode), DELETE otherwise.
463fn fallback_mode(options: &DbOptions) -> JournalSelect {
464    if options.per_host_fallback {
465        JournalSelect::Truncate
466    } else {
467        JournalSelect::Delete
468    }
469}
470
471/// Kill switches (the app's own variable first, then [`JOURNAL_ENV`]); set-but-empty or
472/// unrecognized values are ignored.
473fn env_journal(options: &DbOptions) -> Option<JournalSelect> {
474    options
475        .journal_env
476        .iter()
477        .map(String::as_str)
478        .chain([JOURNAL_ENV])
479        .find_map(|name| {
480            match std::env::var(name)
481                .ok()?
482                .trim()
483                .to_ascii_lowercase()
484                .as_str()
485            {
486                "wal" => Some(JournalSelect::WalStrict),
487                "delete" => Some(JournalSelect::Delete),
488                "truncate" => Some(JournalSelect::Truncate),
489                _ => None,
490            }
491        })
492}
493
494fn effective_journal(options: &DbOptions) -> JournalSelect {
495    if let Some(mode) = env_journal(options) {
496        return mode;
497    }
498    if options.journal == JournalSelect::WalWithFallback
499        && rightkit_fs::is_network_filesystem(&options.path)
500    {
501        return fallback_mode(options);
502    }
503    options.journal
504}
505
506/// The journal mode SQLite is actually running (`wal`, `delete`, ...).
507pub fn journal_mode(conn: &Connection) -> Result<String> {
508    Ok(conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))?)
509}
510
511/// Start a transaction that reserves SQLite's single writer slot up front.
512/// This keeps read-then-write work out of the unretryable BUSY_SNAPSHOT path.
513pub fn write_txn(conn: &mut Connection) -> Result<Transaction<'_>> {
514    conn.transaction_with_behavior(TransactionBehavior::Immediate)
515        .map_err(Into::into)
516}
517
518#[cfg(feature = "cipher")]
519fn apply_key_first(conn: &Connection, key: Option<&str>) -> Result<()> {
520    if let Some(key) = key {
521        let statement = Zeroizing::new(format!("PRAGMA key = {key};"));
522        conn.execute_batch(statement.as_str())?;
523    }
524    Ok(())
525}
526
527#[cfg(not(feature = "cipher"))]
528fn apply_key_first(_conn: &Connection, key: Option<&str>) -> Result<()> {
529    if key.is_some() {
530        return Err(DbError::SqlcipherFeatureRequired);
531    }
532    Ok(())
533}
534
535fn validate_key(key: Option<&str>) -> Result<()> {
536    #[cfg(not(feature = "cipher"))]
537    if key.is_some() {
538        return Err(DbError::SqlcipherFeatureRequired);
539    }
540    if let Some(key) = key {
541        let safe = !key.is_empty()
542            && key.len() <= 4_096
543            && key.bytes().all(|byte| {
544                byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'/' | b'=' | b'-' | b'_')
545            });
546        if !safe {
547            return Err(DbError::InvalidSqlcipherKey);
548        }
549    }
550    Ok(())
551}
552
553fn verify_key(conn: &Connection) -> Result<()> {
554    conn.query_row("SELECT count(*) FROM sqlite_master", [], |row| {
555        row.get::<_, i64>(0)
556    })?;
557    Ok(())
558}
559
560fn integrity_check(conn: &Connection, keyed: bool) -> Result<()> {
561    let pragma = if keyed {
562        "PRAGMA cipher_integrity_check"
563    } else {
564        "PRAGMA integrity_check"
565    };
566    let mut statement = conn.prepare(pragma)?;
567    let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
568    let messages = rows.collect::<std::result::Result<Vec<_>, _>>()?;
569    if messages.is_empty()
570        || messages
571            .iter()
572            .all(|message| message.eq_ignore_ascii_case("ok"))
573    {
574        return Ok(());
575    }
576    Err(DbError::Integrity(messages.join("; ")))
577}
578
579#[derive(Clone)]
580pub struct ReadPool {
581    inner: Arc<ReadPoolInner>,
582}
583
584struct ReadPoolInner {
585    options: DbOptions,
586    cap: usize,
587    state: Mutex<ReadPoolState>,
588    available: Condvar,
589}
590
591#[derive(Default)]
592struct ReadPoolState {
593    idle: Vec<Connection>,
594    total: usize,
595}
596
597pub struct PooledReadConnection {
598    conn: Option<Connection>,
599    pool: Weak<ReadPoolInner>,
600}
601
602impl Deref for PooledReadConnection {
603    type Target = Connection;
604
605    fn deref(&self) -> &Self::Target {
606        self.conn.as_ref().expect("pooled connection is present")
607    }
608}
609
610impl Drop for PooledReadConnection {
611    fn drop(&mut self) {
612        let Some(conn) = self.conn.take() else {
613            return;
614        };
615        let Some(pool) = self.pool.upgrade() else {
616            return;
617        };
618        let mut state = lock_recover(&pool.state);
619        state.idle.push(conn);
620        pool.available.notify_one();
621    }
622}
623
624impl ReadPool {
625    fn open(options: DbOptions, cap: usize) -> Result<Self> {
626        if cap == 0 {
627            return Err(DbError::InvalidPoolCapacity);
628        }
629        let pool = Self {
630            inner: Arc::new(ReadPoolInner {
631                options,
632                cap,
633                state: Mutex::new(ReadPoolState::default()),
634                available: Condvar::new(),
635            }),
636        };
637        let conn = pool.inner.options.open_reader()?;
638        let mut state = lock_recover(&pool.inner.state);
639        state.total = 1;
640        state.idle.push(conn);
641        drop(state);
642        Ok(pool)
643    }
644
645    pub fn get(&self) -> Result<PooledReadConnection> {
646        let mut state = lock_recover(&self.inner.state);
647        loop {
648            if let Some(conn) = state.idle.pop() {
649                return Ok(PooledReadConnection {
650                    conn: Some(conn),
651                    pool: Arc::downgrade(&self.inner),
652                });
653            }
654            if state.total < self.inner.cap {
655                state.total += 1;
656                drop(state);
657                return match self.inner.options.open_reader() {
658                    Ok(conn) => Ok(PooledReadConnection {
659                        conn: Some(conn),
660                        pool: Arc::downgrade(&self.inner),
661                    }),
662                    Err(error) => {
663                        let mut state = lock_recover(&self.inner.state);
664                        state.total -= 1;
665                        self.inner.available.notify_one();
666                        Err(error)
667                    }
668                };
669            }
670            state = match self.inner.available.wait(state) {
671                Ok(state) => state,
672                Err(poisoned) => poisoned.into_inner(),
673            };
674        }
675    }
676
677    pub fn capacity(&self) -> usize {
678        self.inner.cap
679    }
680
681    pub fn idle_len(&self) -> usize {
682        lock_recover(&self.inner.state).idle.len()
683    }
684
685    pub fn total_len(&self) -> usize {
686        lock_recover(&self.inner.state).total
687    }
688}
689
690fn lock_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
691    mutex
692        .lock()
693        .unwrap_or_else(|poisoned| poisoned.into_inner())
694}
695
696#[derive(Clone, Copy, Debug, Eq, PartialEq)]
697pub struct CheckpointStatus {
698    pub busy: i64,
699    pub log_frames: i64,
700    pub checkpointed_frames: i64,
701}
702
703pub fn checkpoint_truncate(conn: &Connection) -> Result<CheckpointStatus> {
704    checkpoint(conn, "PRAGMA wal_checkpoint(TRUNCATE)")
705}
706
707pub fn checkpoint_if_wal_larger_than(
708    conn: &Connection,
709    db: &Path,
710    bytes: u64,
711) -> Result<Option<CheckpointStatus>> {
712    let wal = wal_path(db)?;
713    let size = match std::fs::metadata(wal) {
714        Ok(metadata) => metadata.len(),
715        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
716        Err(_) => return Ok(None),
717    };
718    if size <= bytes {
719        return Ok(None);
720    }
721    checkpoint(conn, "PRAGMA wal_checkpoint(PASSIVE)").map(Some)
722}
723
724fn checkpoint(conn: &Connection, pragma: &str) -> Result<CheckpointStatus> {
725    conn.query_row(pragma, [], |row| {
726        Ok(CheckpointStatus {
727            busy: row.get(0)?,
728            log_frames: row.get(1)?,
729            checkpointed_frames: row.get(2)?,
730        })
731    })
732    .map_err(Into::into)
733}
734
735fn wal_path(db: &Path) -> Result<PathBuf> {
736    let file_name = db.file_name().ok_or(DbError::InvalidDatabasePath)?;
737    let mut wal_name = OsString::from(file_name);
738    wal_name.push("-wal");
739    Ok(db.with_file_name(wal_name))
740}
741
742pub fn optimize_on_open(conn: &Connection) -> Result<()> {
743    conn.execute_batch("PRAGMA optimize=0x10002;")?;
744    Ok(())
745}
746
747pub fn optimize_boundary(conn: &Connection) -> Result<()> {
748    conn.execute_batch("PRAGMA optimize;")?;
749    Ok(())
750}
rightkit_db/lib.rs

rightkit_db/
lib.rs
