rightkit_db/
lib.rs

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

rightkit_db/
lib.rs
