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, Instant};
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;
25mod unreadable;
26
27pub use backup::{online_backup_to, restore_from, BackupProgress};
28pub use migrate::{
29    add_column_if_missing, column_exists, migrate, migration_version, pending_marker, rollback_to,
30    rollback_to_with_storage, user_version, LegacyVersionTable, MigrateOptions, Migration,
31    MigrationReport, PendingMarker, Step, VersionStorage,
32};
33#[cfg(feature = "cipher")]
34pub use recover::rekey;
35pub use recover::{
36    quarantine_database, quarantine_database_opts, quarantine_database_with, MoveMode, OpenOutcome,
37    QuarantineOptions, QuarantineStamp, RecoveryPolicy, SidecarPolicy,
38};
39
40pub use unreadable::{
41    classify_open_failure, unreadable_counter_path, RecoveryAction, StorageProblem,
42    UnreadableCause, UnreadablePolicy, UnreadableRecovery,
43};
44
45const BUSY_TIMEOUT_MS: u64 = 5_000;
46const JOURNAL_SIZE_LIMIT: i64 = 67_108_864;
47
48pub type Result<T> = std::result::Result<T, DbError>;
49
50#[derive(Debug, thiserror::Error)]
51#[non_exhaustive]
52pub enum DbError {
53    #[error("SQLite error: {0}")]
54    Sqlite(#[from] rusqlite::Error),
55    #[error("a SQLCipher key requires the rightkit-db `sqlcipher` or `sqlcipher-system` feature")]
56    SqlcipherFeatureRequired,
57    #[error("SQLCipher key contains characters that are unsafe for PRAGMA interpolation")]
58    InvalidSqlcipherKey,
59    /// An open-time `quick_check` or `integrity_check` failed. Open-time recovery treats it as an
60    /// unreadable database (see `DbOptions::open_writer_recovering`).
61    #[error("database integrity check failed: {0}")]
62    Integrity(String),
63    #[error("quarantine suffix {0:?} must be non-empty and contain no path separator or NUL")]
64    InvalidQuarantineSuffix(String),
65    #[error("read-pool capacity must be greater than zero")]
66    InvalidPoolCapacity,
67    #[error("reader cache size must be between 1 and 2147483647 KiB, got {0}")]
68    InvalidReaderCacheSize(u32),
69    /// Checkout after [`ReadPool::close`] or [`ReadPool::close_and_wait`].
70    #[error("read pool is closed")]
71    ReadPoolClosed,
72    /// [`ReadPool::close_and_wait`] timed out. `outstanding` counts readers still checked out or
73    /// opening.
74    #[error("read pool close timed out after {timeout:?} with {outstanding} reader(s) still out")]
75    ReadPoolCloseTimeout {
76        outstanding: usize,
77        timeout: Duration,
78    },
79    /// A reader did not come up read-only: `PRAGMA query_only` did not stay on, or a read-only
80    /// handle reported a writable main database.
81    #[error("reader connection is not read-only: {0}")]
82    ReadOnlyNotEffective(String),
83    #[error("database path has no file name")]
84    InvalidDatabasePath,
85    #[error("database is at schema v{current}, newer than this build understands (v{max}) and not forward compatible")]
86    NewerDatabase { current: i64, max: i64 },
87    #[error("invalid migration list: {0}")]
88    InvalidMigrations(String),
89    #[error("migration v{version} ({name}) failed and was rolled back: {source}")]
90    MigrationFailed {
91        version: i64,
92        name: &'static str,
93        #[source]
94        source: rusqlite::Error,
95    },
96    #[error("cannot roll back: {0}")]
97    Rollback(String),
98    #[error("database backup failed: {0}")]
99    Backup(String),
100    #[error("no candidate key opened the database")]
101    NoKeyOpened,
102    #[error(transparent)]
103    Fs(#[from] rightkit_fs::FsError),
104    /// The database file could not be read (permission or I/O failure) or its storage is
105    /// unavailable. Not corruption: the file was not touched unless `quarantine_after` was
106    /// reached, in which case recovery succeeds instead of returning this. See
107    /// [`UnreadablePolicy`].
108    #[error(
109        "database {} cannot be read: {cause} ({consecutive_failures} consecutive failed open(s)): {source}",
110        .path.display()
111    )]
112    Unreadable {
113        path: PathBuf,
114        cause: UnreadableCause,
115        /// Consecutive failed opens recorded so far (0 under [`UnreadablePolicy::Fail`] and for
116        /// storage problems, which are never counted).
117        consecutive_failures: u32,
118        /// The count at which [`UnreadablePolicy::QuarantineAfter`] would set the file aside.
119        quarantine_after: Option<u32>,
120        #[source]
121        source: Box<DbError>,
122    },
123}
124
125#[derive(Clone, Copy, Debug, Eq, PartialEq)]
126pub enum CacheProfile {
127    Default,
128    LargeWorkingSet,
129}
130
131#[derive(Clone, Copy, Debug, Eq, PartialEq)]
132pub enum DurabilityProfile {
133    Full,
134    Normal,
135}
136
137/// How the journal mode is chosen. WAL needs working shared-memory locking, which network
138/// filesystems often lack and which SQLite will silently downgrade; the default therefore
139/// *verifies* the mode that took effect and falls back to a rollback journal.
140#[derive(Clone, Copy, Debug, Eq, PartialEq)]
141#[non_exhaustive]
142pub enum JournalSelect {
143    /// WAL when the filesystem honors it and is not a detected network mount; else a
144    /// rollback journal (TRUNCATE with [`DbOptions::per_host_fallback`], else DELETE).
145    WalWithFallback,
146    /// WAL or fail (no fallback).
147    WalStrict,
148    /// Rollback journal, deleted after each transaction.
149    Delete,
150    /// Rollback journal, truncated after each transaction.
151    Truncate,
152}
153
154impl JournalSelect {
155    fn is_rollback(self) -> bool {
156        matches!(self, JournalSelect::Delete | JournalSelect::Truncate)
157    }
158}
159
160/// Environment kill-switch (`wal` | `delete` | `truncate`) for a misclassified mount in the
161/// field. An app's own variable ([`DbOptions::journal_env`]) is checked first.
162pub const JOURNAL_ENV: &str = "RIGHTKIT_SQLITE_JOURNAL_MODE";
163
164/// Post-key, pre-pragma per-connection initializer (register functions, collations, ...).
165/// Open-time integrity check, selected by [`DbOptions::integrity`].
166#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
167pub enum IntegrityCheck {
168    /// No check at open (default).
169    #[default]
170    Off,
171    /// `PRAGMA quick_check`: structural page checks, without the index-to-table cross-check. It
172    /// costs far less than `Full` on large databases.
173    Quick,
174    /// `PRAGMA integrity_check`, or `PRAGMA cipher_integrity_check` for a keyed database.
175    Full,
176}
177
178pub type ConnInit = Arc<dyn Fn(&Connection) -> rusqlite::Result<()> + Send + Sync>;
179
180#[derive(Clone)]
181pub struct DbOptions {
182    pub path: PathBuf,
183    /// SQLite busy wait, installed before key verification or any bootstrap statement.
184    /// Default: five seconds.
185    pub busy_timeout: Duration,
186    /// SQLCipher passphrase-style key (base64-ish charset). See also [`DbOptions::raw_key`].
187    pub key: Option<Zeroizing<String>>,
188    /// SQLCipher raw 256-bit key, applied as `x'<hex>'` (no KDF). Takes precedence over `key`.
189    pub raw_key: Option<Zeroizing<[u8; 32]>>,
190    pub profile: CacheProfile,
191    pub durability: DurabilityProfile,
192    pub foreign_keys: bool,
193    /// Legacy switch: `true` runs a full check on every open. Ignored when `integrity` is not `Off`.
194    pub integrity_check_on_open: bool,
195    /// Open-time integrity check (default `Off`). A failed check is [`DbError::Integrity`], which
196    /// `open_writer_recovering` treats as unreadable.
197    pub integrity: IntegrityCheck,
198    pub journal: JournalSelect,
199    /// Narrow the database and its sidecars to owner-only (0600) on unix. Default on. Set it to
200    /// `false` when sibling processes must read the file (for example a helper running as another
201    /// user); that is the intended switch, not a bug.
202    pub owner_only: bool,
203    pub init: Option<ConnInit>,
204    /// The app's own kill-switch variable (e.g. `CODERIGHT_SQLITE_JOURNAL_MODE`), checked
205    /// before [`JOURNAL_ENV`].
206    pub journal_env: Option<String>,
207    /// When a rollback journal is in force (network mount, kill switch, or WAL refused),
208    /// open the per-host sibling [`per_host_path`] instead of the requested file, so hosts
209    /// sharing a network directory never share one rollback-journaled database. The
210    /// requested file is not created. Default off.
211    pub per_host_fallback: bool,
212    /// Sets `PRAGMA secure_delete` on every connection at open: `Some(true)` overwrites freed
213    /// pages, `Some(false)` does not, `None` (default) leaves SQLite's compile-time default.
214    pub secure_delete: Option<bool>,
215    /// Run `PRAGMA optimize` on each writer connection at open (default `true`). Set it to `false`
216    /// to leave planner statistics alone at open, for example when the file is opened only to
217    /// inspect it. Readers never run it, whatever this is set to.
218    pub optimize_on_open: bool,
219}
220
221impl DbOptions {
222    pub fn new(path: impl Into<PathBuf>) -> Self {
223        Self {
224            path: path.into(),
225            busy_timeout: Duration::from_millis(BUSY_TIMEOUT_MS),
226            key: None,
227            raw_key: None,
228            profile: CacheProfile::Default,
229            durability: DurabilityProfile::Normal,
230            foreign_keys: false,
231            integrity_check_on_open: false,
232            integrity: IntegrityCheck::Off,
233            journal: JournalSelect::WalWithFallback,
234            owner_only: true,
235            init: None,
236            journal_env: None,
237            per_host_fallback: false,
238            secure_delete: None,
239            optimize_on_open: true,
240        }
241    }
242
243    pub fn with_journal_env(mut self, name: impl Into<String>) -> Self {
244        self.journal_env = Some(name.into());
245        self
246    }
247
248    pub fn with_per_host_fallback(mut self, on: bool) -> Self {
249        self.per_host_fallback = on;
250        self
251    }
252
253    /// Turn `PRAGMA optimize` off at writer open (see [`DbOptions::optimize_on_open`]).
254    pub fn without_optimize_on_open(mut self) -> Self {
255        self.optimize_on_open = false;
256        self
257    }
258
259    /// The file connections actually open: the requested path, or its per-host sibling when
260    /// [`Self::per_host_fallback`] applies (a rollback journal is in force, or a writer
261    /// already fell back there).
262    pub fn effective_path(&self) -> PathBuf {
263        if !self.per_host_fallback {
264            return self.path.clone();
265        }
266        let sibling = per_host_path(&self.path);
267        if effective_journal(self).is_rollback() || (sibling.exists() && !self.path.exists()) {
268            sibling
269        } else {
270            self.path.clone()
271        }
272    }
273
274    pub fn with_passphrase(mut self, key: impl Into<String>) -> Self {
275        self.key = Some(Zeroizing::new(key.into()));
276        self.raw_key = None;
277        self
278    }
279
280    pub fn with_raw_key(mut self, key: [u8; 32]) -> Self {
281        self.raw_key = Some(Zeroizing::new(key));
282        self.key = None;
283        self
284    }
285
286    pub fn is_keyed(&self) -> bool {
287        self.key.is_some() || self.raw_key.is_some()
288    }
289
290    pub fn open_writer(&self) -> Result<Connection> {
291        self.open(ConnectionKind::Writer)
292    }
293
294    pub fn open_reader(&self) -> Result<Connection> {
295        self.open(ConnectionKind::Reader { cache_kib: None })
296    }
297
298    /// Open exactly `path` read-only, without creating it or running mutating bootstrap.
299    /// URI parameters are passed to SQLite unchanged. Journal fallback, permissions,
300    /// `init`, optimization, and migrations are skipped; key and timeout still apply.
301    pub fn open_read_only(&self) -> Result<Connection> {
302        self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_ONLY)
303    }
304
305    /// Open exactly `path` read-write without CREATE or mutating bootstrap.
306    /// Uses the same minimal setup as [`Self::open_read_only`].
307    pub fn open_existing_writer(&self) -> Result<Connection> {
308        self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_WRITE)
309    }
310
311    /// Open a private, ephemeral database. `path`, filesystem permissions, journal
312    /// selection/fallback, and mmap are ignored; other bootstrap options still apply.
313    pub fn open_in_memory(&self) -> Result<Connection> {
314        let key = self.key_pragma_value()?;
315        let conn = Connection::open_in_memory()?;
316        self.initialize(&conn, key.as_ref().map(|k| k.as_str()))?;
317        configure_connection(&conn, self, ConnectionKind::Memory, None)?;
318        self.check_on_open(&conn)?;
319        Ok(conn)
320    }
321
322    fn open_existing_at(&self, path: &Path, access: OpenFlags) -> Result<Connection> {
323        let key = self.key_pragma_value()?;
324        let conn = Connection::open_with_flags(
325            path,
326            access | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX,
327        )?;
328        conn.busy_timeout(self.busy_timeout)?;
329        apply_key_first(&conn, key.as_ref().map(|k| k.as_str()))?;
330        verify_key(&conn)?;
331        self.check_on_open(&conn)?;
332        Ok(conn)
333    }
334
335    fn initialize(&self, conn: &Connection, key: Option<&str>) -> Result<()> {
336        conn.busy_timeout(self.busy_timeout)?;
337        apply_key_first(conn, key)?;
338        verify_key(conn)?;
339        if let Some(init) = &self.init {
340            init(conn)?;
341        }
342        Ok(())
343    }
344
345    /// A read pool with [`ReadPoolOptions::new`]: read-write file handles, with `query_only` set
346    /// and verified on every reader. Same as `open_read_pool_with(cap, ReadPoolOptions::new())`.
347    pub fn open_read_pool(&self, cap: usize) -> Result<ReadPool> {
348        ReadPool::open(self.clone(), cap, ReadPoolOptions::new())
349    }
350
351    /// A read pool configured by [`ReadPoolOptions`]. A read-only pool needs the database to exist.
352    pub fn open_read_pool_with(&self, cap: usize, pool: ReadPoolOptions) -> Result<ReadPool> {
353        ReadPool::open(self.clone(), cap, pool)
354    }
355
356    fn open_pool_reader(&self, pool: ReadPoolOptions) -> Result<Connection> {
357        if pool.read_only {
358            self.open_pool_read_only(pool.cache_kib)
359        } else {
360            self.open(ConnectionKind::Reader {
361                cache_kib: pool.cache_kib,
362            })
363        }
364    }
365
366    /// A pool reader opened with `SQLITE_OPEN_READ_ONLY`, so the file refuses writes even after
367    /// `query_only` is turned off. It never creates the file, changes the journal mode, narrows
368    /// permissions, or runs `optimize` or `init`; the writer owns those.
369    fn open_pool_read_only(&self, cache_kib: Option<u32>) -> Result<Connection> {
370        let path = self.effective_path();
371        let key = self.key_pragma_value()?;
372        let conn = Connection::open_with_flags(
373            &path,
374            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
375        )?;
376        conn.busy_timeout(self.busy_timeout)?;
377        apply_key_first(&conn, key.as_ref().map(|k| k.as_str()))?;
378        verify_key(&conn)?;
379        conn.pragma_update(None, "temp_store", "MEMORY")?;
380        if self.foreign_keys {
381            conn.pragma_update(None, "foreign_keys", true)?;
382        }
383        apply_cache_settings(&conn, self, ConnectionKind::Reader { cache_kib })?;
384        enforce_query_only(&conn)?;
385        if !conn.is_readonly("main")? {
386            return Err(DbError::ReadOnlyNotEffective(
387                "SQLITE_OPEN_READ_ONLY handle reports a writable main database".into(),
388            ));
389        }
390        self.check_on_open(&conn)?;
391        Ok(conn)
392    }
393
394    fn open(&self, kind: ConnectionKind) -> Result<Connection> {
395        let path = self.effective_path();
396        let existed = path.exists();
397        // The per-host sibling is always rollback-journaled, whoever opens it.
398        let forced = (path != self.path).then(|| match effective_journal(self) {
399            mode if mode.is_rollback() => mode,
400            _ => fallback_mode(self),
401        });
402        let (conn, journal) = self.open_at(&path, kind, forced)?;
403        // WAL refused at runtime: with per-host fallback, the rollback journal belongs on the
404        // per-host sibling, and a file this open created at the requested path goes away.
405        if journal == Opened::FellBack && self.per_host_fallback && path == self.path {
406            drop(conn);
407            if !existed {
408                for suffix in ["", "-journal", "-wal", "-shm"] {
409                    let mut name = OsString::from(path.as_os_str());
410                    name.push(suffix);
411                    let _ = std::fs::remove_file(PathBuf::from(name));
412                }
413            }
414            let sibling = per_host_path(&self.path);
415            return Ok(self
416                .open_at(&sibling, kind, Some(JournalSelect::Truncate))?
417                .0);
418        }
419        Ok(conn)
420    }
421
422    fn open_at(
423        &self,
424        path: &Path,
425        kind: ConnectionKind,
426        forced: Option<JournalSelect>,
427    ) -> Result<(Connection, Opened)> {
428        let key = self.key_pragma_value()?;
429        let flags = OpenFlags::SQLITE_OPEN_READ_WRITE
430            | OpenFlags::SQLITE_OPEN_CREATE
431            | OpenFlags::SQLITE_OPEN_NO_MUTEX;
432        let conn = Connection::open_with_flags(path, flags)?;
433        // Narrow the main file before any write so WAL/SHM sidecars inherit owner-only.
434        if self.owner_only {
435            rightkit_fs::restrict_owner_only(path)?;
436        }
437        self.initialize(&conn, key.as_ref().map(|k| k.as_str()))?;
438        if let Some(on) = self.secure_delete {
439            conn.pragma_update(None, "secure_delete", if on { "ON" } else { "OFF" })?;
440        }
441        let journal = configure_connection(&conn, self, kind, forced)?;
442        self.check_on_open(&conn)?;
443        if self.owner_only {
444            harden_sidecars(path)?;
445        }
446        Ok((conn, journal))
447    }
448
449    /// Run the open-time check selected by `integrity`; `integrity_check_on_open` alone means `Full`.
450    fn check_on_open(&self, conn: &Connection) -> Result<()> {
451        let mode = match self.integrity {
452            IntegrityCheck::Off if self.integrity_check_on_open => IntegrityCheck::Full,
453            mode => mode,
454        };
455        match (mode, self.is_keyed()) {
456            (IntegrityCheck::Off, _) => Ok(()),
457            (IntegrityCheck::Quick, _) => run_integrity_pragma(conn, "PRAGMA quick_check"),
458            (IntegrityCheck::Full, true) => {
459                run_integrity_pragma(conn, "PRAGMA cipher_integrity_check")
460            }
461            (IntegrityCheck::Full, false) => run_integrity_pragma(conn, "PRAGMA integrity_check"),
462        }
463    }
464
465    /// The `PRAGMA key` right-hand side (already quoted), validated against injection.
466    fn key_pragma_value(&self) -> Result<Option<Zeroizing<String>>> {
467        if let Some(raw) = &self.raw_key {
468            #[cfg(not(feature = "cipher"))]
469            {
470                let _ = raw;
471                return Err(DbError::SqlcipherFeatureRequired);
472            }
473            #[cfg(feature = "cipher")]
474            {
475                let hex: String = raw.iter().map(|b| format!("{b:02x}")).collect();
476                return Ok(Some(Zeroizing::new(format!("\"x'{hex}'\""))));
477            }
478        }
479        let key = self.key.as_ref().map(|key| key.as_str());
480        validate_key(key)?;
481        Ok(key.map(|k| Zeroizing::new(format!("'{k}'"))))
482    }
483}
484
485/// Default-options read-only open; never creates a missing database.
486pub fn open_read_only(path: impl AsRef<Path>) -> Result<Connection> {
487    DbOptions::new(path.as_ref()).open_read_only()
488}
489
490/// Default-options read-write open; never creates a missing database.
491pub fn open_existing_writer(path: impl AsRef<Path>) -> Result<Connection> {
492    DbOptions::new(path.as_ref()).open_existing_writer()
493}
494
495/// Default-options private in-memory database.
496pub fn open_in_memory() -> Result<Connection> {
497    DbOptions::new(":memory:").open_in_memory()
498}
499
500fn harden_sidecars(db: &Path) -> Result<()> {
501    for suffix in ["-wal", "-shm"] {
502        let mut name = OsString::from(db.file_name().ok_or(DbError::InvalidDatabasePath)?);
503        name.push(suffix);
504        rightkit_fs::restrict_owner_only(&db.with_file_name(name))?;
505    }
506    Ok(())
507}
508
509#[derive(Clone, Copy)]
510enum ConnectionKind {
511    Writer,
512    /// `cache_kib` overrides the reader cache size; see [`ReadPoolOptions::with_reader_cache_kib`].
513    Reader {
514        cache_kib: Option<u32>,
515    },
516    Memory,
517}
518
519impl ConnectionKind {
520    fn reader_cache_kib(self) -> Option<u32> {
521        match self {
522            ConnectionKind::Reader { cache_kib } => cache_kib,
523            ConnectionKind::Writer | ConnectionKind::Memory => None,
524        }
525    }
526}
527
528/// What journal setup did.
529#[derive(Clone, Copy, Debug, Eq, PartialEq)]
530enum Opened {
531    AsRequested,
532    /// WAL was wanted but not honored; a rollback journal is in force.
533    FellBack,
534}
535
536fn configure_connection(
537    conn: &Connection,
538    options: &DbOptions,
539    kind: ConnectionKind,
540    forced: Option<JournalSelect>,
541) -> Result<Opened> {
542    let opened = if matches!(kind, ConnectionKind::Memory) {
543        Opened::AsRequested
544    } else {
545        configure_journal(conn, options, kind, forced)?
546    };
547    let synchronous = match options.durability {
548        DurabilityProfile::Full => "FULL",
549        DurabilityProfile::Normal => "NORMAL",
550    };
551    conn.pragma_update(None, "synchronous", synchronous)?;
552    conn.pragma_update(None, "journal_size_limit", JOURNAL_SIZE_LIMIT)?;
553    conn.pragma_update(None, "temp_store", "MEMORY")?;
554    if options.foreign_keys {
555        conn.pragma_update(None, "foreign_keys", true)?;
556    }
557    apply_cache_settings(conn, options, kind)?;
558    if matches!(kind, ConnectionKind::Reader { .. }) {
559        // Readers never run `PRAGMA optimize`: it can write statistics, and the writer runs it
560        // when it opens. `query_only` is set after every other reader setup statement.
561        enforce_query_only(conn)?;
562    } else if options.optimize_on_open {
563        optimize_on_open(conn)?;
564    }
565    Ok(opened)
566}
567
568/// Profile cache and mmap sizes, then the reader override, which wins over the profile on readers
569/// only. Writers and in-memory connections keep the profile size.
570fn apply_cache_settings(
571    conn: &Connection,
572    options: &DbOptions,
573    kind: ConnectionKind,
574) -> Result<()> {
575    if options.profile == CacheProfile::LargeWorkingSet {
576        conn.pragma_update(None, "cache_size", -65_536_i64)?;
577        if !matches!(kind, ConnectionKind::Memory) {
578            conn.pragma_update(None, "mmap_size", 268_435_456_i64)?;
579        }
580    }
581    if let Some(kib) = kind.reader_cache_kib() {
582        conn.pragma_update(None, "cache_size", -i64::from(kib))?;
583    }
584    Ok(())
585}
586
587/// Sets `PRAGMA query_only = ON` on a reader and reads it back, so a reader that did not come up
588/// query-only is never handed out.
589fn enforce_query_only(conn: &Connection) -> Result<()> {
590    conn.pragma_update(None, "query_only", true)?;
591    let on: i64 = conn.query_row("PRAGMA query_only", [], |row| row.get(0))?;
592    if on != 1 {
593        return Err(DbError::ReadOnlyNotEffective(
594            "PRAGMA query_only did not stay ON".into(),
595        ));
596    }
597    Ok(())
598}
599
600fn set_rollback(conn: &Connection, mode: JournalSelect) -> Result<()> {
601    let pragma = if mode == JournalSelect::Truncate {
602        "PRAGMA journal_mode=TRUNCATE"
603    } else {
604        "PRAGMA journal_mode=DELETE"
605    };
606    conn.query_row(pragma, [], |row| row.get::<_, String>(0))?;
607    Ok(())
608}
609
610fn configure_journal(
611    conn: &Connection,
612    options: &DbOptions,
613    kind: ConnectionKind,
614    forced: Option<JournalSelect>,
615) -> Result<Opened> {
616    let wanted = forced.unwrap_or_else(|| effective_journal(options));
617    if wanted.is_rollback() {
618        // Rollback modes are per connection: a reader left in the default DELETE mode would
619        // try to delete a TRUNCATE writer's open journal (IOERR_DELETE on Windows).
620        set_rollback(conn, wanted)?;
621        return Ok(Opened::AsRequested);
622    }
623    let honored = conn
624        .query_row("PRAGMA journal_mode=WAL", [], |row| row.get::<_, String>(0))
625        .map(|mode| mode.eq_ignore_ascii_case("wal"))
626        .unwrap_or(false)
627        // `journal_mode=WAL` can report success before the -shm/-wal sidecars are usable;
628        // one empty write transaction proves the sidecars work before anything depends on them.
629        && (matches!(kind, ConnectionKind::Reader { .. })
630            || conn.execute_batch("BEGIN IMMEDIATE; ROLLBACK;").is_ok());
631    if honored {
632        return Ok(Opened::AsRequested);
633    }
634    if wanted == JournalSelect::WalStrict {
635        return Err(DbError::Sqlite(rusqlite::Error::SqliteFailure(
636            rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR),
637            Some("WAL journal mode was not honored by this filesystem".into()),
638        )));
639    }
640    let _ = conn.execute_batch("ROLLBACK;");
641    set_rollback(conn, fallback_mode(options))?;
642    Ok(Opened::FellBack)
643}
644
645/// The rollback journal used when WAL is unavailable: TRUNCATE with per-host fallback
646/// (CodeRight's network mode), DELETE otherwise.
647fn fallback_mode(options: &DbOptions) -> JournalSelect {
648    if options.per_host_fallback {
649        JournalSelect::Truncate
650    } else {
651        JournalSelect::Delete
652    }
653}
654
655/// Kill switches (the app's own variable first, then [`JOURNAL_ENV`]); set-but-empty or
656/// unrecognized values are ignored.
657fn env_journal(options: &DbOptions) -> Option<JournalSelect> {
658    options
659        .journal_env
660        .iter()
661        .map(String::as_str)
662        .chain([JOURNAL_ENV])
663        .find_map(|name| {
664            match std::env::var(name)
665                .ok()?
666                .trim()
667                .to_ascii_lowercase()
668                .as_str()
669            {
670                "wal" => Some(JournalSelect::WalStrict),
671                "delete" => Some(JournalSelect::Delete),
672                "truncate" => Some(JournalSelect::Truncate),
673                _ => None,
674            }
675        })
676}
677
678fn effective_journal(options: &DbOptions) -> JournalSelect {
679    if let Some(mode) = env_journal(options) {
680        return mode;
681    }
682    if options.journal == JournalSelect::WalWithFallback
683        && rightkit_fs::is_network_filesystem(&options.path)
684    {
685        return fallback_mode(options);
686    }
687    options.journal
688}
689
690/// The journal mode SQLite is actually running (`wal`, `delete`, ...).
691pub fn journal_mode(conn: &Connection) -> Result<String> {
692    Ok(conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))?)
693}
694
695/// Start a transaction that reserves SQLite's single writer slot up front.
696/// This keeps read-then-write work out of the unretryable BUSY_SNAPSHOT path.
697pub fn write_txn(conn: &mut Connection) -> Result<Transaction<'_>> {
698    conn.transaction_with_behavior(TransactionBehavior::Immediate)
699        .map_err(Into::into)
700}
701
702#[cfg(feature = "cipher")]
703fn apply_key_first(conn: &Connection, key: Option<&str>) -> Result<()> {
704    if let Some(key) = key {
705        let statement = Zeroizing::new(format!("PRAGMA key = {key};"));
706        conn.execute_batch(statement.as_str())?;
707    }
708    Ok(())
709}
710
711#[cfg(not(feature = "cipher"))]
712fn apply_key_first(_conn: &Connection, key: Option<&str>) -> Result<()> {
713    if key.is_some() {
714        return Err(DbError::SqlcipherFeatureRequired);
715    }
716    Ok(())
717}
718
719fn validate_key(key: Option<&str>) -> Result<()> {
720    #[cfg(not(feature = "cipher"))]
721    if key.is_some() {
722        return Err(DbError::SqlcipherFeatureRequired);
723    }
724    if let Some(key) = key {
725        let safe = !key.is_empty()
726            && key.len() <= 4_096
727            && key.bytes().all(|byte| {
728                byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'/' | b'=' | b'-' | b'_')
729            });
730        if !safe {
731            return Err(DbError::InvalidSqlcipherKey);
732        }
733    }
734    Ok(())
735}
736
737fn verify_key(conn: &Connection) -> Result<()> {
738    conn.query_row("SELECT count(*) FROM sqlite_master", [], |row| {
739        row.get::<_, i64>(0)
740    })?;
741    Ok(())
742}
743
744fn run_integrity_pragma(conn: &Connection, pragma: &str) -> Result<()> {
745    let mut statement = conn.prepare(pragma)?;
746    let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
747    let messages = rows.collect::<std::result::Result<Vec<_>, _>>()?;
748    if messages.is_empty()
749        || messages
750            .iter()
751            .all(|message| message.eq_ignore_ascii_case("ok"))
752    {
753        return Ok(());
754    }
755    Err(DbError::Integrity(messages.join("; ")))
756}
757
758/// Options for a [`ReadPool`]. The default ([`ReadPoolOptions::new`]) matches
759/// [`DbOptions::open_read_pool`]: read-write file handles, with `query_only` set and verified on
760/// every reader. Build with the `with_*` methods; the fields are private.
761#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
762pub struct ReadPoolOptions {
763    read_only: bool,
764    cache_kib: Option<u32>,
765}
766
767impl ReadPoolOptions {
768    pub fn new() -> Self {
769        Self::default()
770    }
771
772    /// Opens each reader with `SQLITE_OPEN_READ_ONLY`, so writes fail even if `query_only` is later
773    /// turned off. The database must already exist: the pool never creates it, changes its journal
774    /// mode, narrows its permissions, or runs `optimize` or `init`. A rollback journal left by a
775    /// crash is recovered by a writer, not by a read-only reader. Default off.
776    pub fn with_read_only(mut self, on: bool) -> Self {
777        self.read_only = on;
778        self
779    }
780
781    /// Page cache size for each reader connection, in KiB. It is applied as `PRAGMA cache_size`
782    /// set to `-kib`, so the size does not depend on the page size. Valid values are 1 to
783    /// 2147483647; the pool checks this when it opens. It overrides
784    /// [`CacheProfile::LargeWorkingSet`] on readers only. Writers and [`DbOptions::open_reader`]
785    /// keep the profile size. Default: the profile's size.
786    pub fn with_reader_cache_kib(mut self, kib: u32) -> Self {
787        self.cache_kib = Some(kib);
788        self
789    }
790}
791
792/// A bounded pool of reader connections. Clones share one pool.
793#[derive(Clone)]
794pub struct ReadPool {
795    inner: Arc<ReadPoolInner>,
796}
797
798struct ReadPoolInner {
799    options: DbOptions,
800    pool_options: ReadPoolOptions,
801    cap: usize,
802    state: Mutex<ReadPoolState>,
803    /// Wakes a `get` waiting for capacity, and every waiter when the pool closes.
804    available: Condvar,
805    /// Wakes `close_and_wait` when a connection is released after the pool closed.
806    drained: Condvar,
807}
808
809#[derive(Default)]
810struct ReadPoolState {
811    idle: Vec<Connection>,
812    /// Idle, checked-out, and still-opening connections.
813    total: usize,
814    closed: bool,
815}
816
817impl ReadPoolInner {
818    fn open_reader(&self) -> Result<Connection> {
819        self.options.open_pool_reader(self.pool_options)
820    }
821}
822
823pub struct PooledReadConnection {
824    conn: Option<Connection>,
825    pool: Weak<ReadPoolInner>,
826}
827
828impl Deref for PooledReadConnection {
829    type Target = Connection;
830
831    fn deref(&self) -> &Self::Target {
832        self.conn.as_ref().expect("pooled connection is present")
833    }
834}
835
836impl Drop for PooledReadConnection {
837    fn drop(&mut self) {
838        let Some(conn) = self.conn.take() else {
839            return;
840        };
841        let Some(pool) = self.pool.upgrade() else {
842            return;
843        };
844        let mut state = lock_recover(&pool.state);
845        if state.closed {
846            // Close the file before the count drops, so `close_and_wait` returns only once it is
847            // released. Checked under the same lock as `close`, so no connection is parked idle
848            // after the pool has closed.
849            drop(conn);
850            state.total -= 1;
851            pool.drained.notify_all();
852        } else {
853            state.idle.push(conn);
854            pool.available.notify_one();
855        }
856    }
857}
858
859impl ReadPool {
860    fn open(options: DbOptions, cap: usize, pool_options: ReadPoolOptions) -> Result<Self> {
861        if cap == 0 {
862            return Err(DbError::InvalidPoolCapacity);
863        }
864        if let Some(kib) = pool_options.cache_kib {
865            if kib == 0 || kib > i32::MAX.unsigned_abs() {
866                return Err(DbError::InvalidReaderCacheSize(kib));
867            }
868        }
869        let pool = Self {
870            inner: Arc::new(ReadPoolInner {
871                options,
872                pool_options,
873                cap,
874                state: Mutex::new(ReadPoolState::default()),
875                available: Condvar::new(),
876                drained: Condvar::new(),
877            }),
878        };
879        let conn = pool.inner.open_reader()?;
880        let mut state = lock_recover(&pool.inner.state);
881        state.total = 1;
882        state.idle.push(conn);
883        drop(state);
884        Ok(pool)
885    }
886
887    /// Checks out an idle reader, opens a new one while under the cap, or waits for a return.
888    /// Fails with [`DbError::ReadPoolClosed`] once the pool is closed, including while waiting.
889    pub fn get(&self) -> Result<PooledReadConnection> {
890        let mut state = lock_recover(&self.inner.state);
891        loop {
892            if state.closed {
893                return Err(DbError::ReadPoolClosed);
894            }
895            if let Some(conn) = state.idle.pop() {
896                return Ok(PooledReadConnection {
897                    conn: Some(conn),
898                    pool: Arc::downgrade(&self.inner),
899                });
900            }
901            if state.total < self.inner.cap {
902                state.total += 1;
903                drop(state);
904                return self.open_checked_out();
905            }
906            state = match self.inner.available.wait(state) {
907                Ok(state) => state,
908                Err(poisoned) => poisoned.into_inner(),
909            };
910        }
911    }
912
913    /// Opens a reader for a `get` that reserved a slot. If the pool closed meanwhile, the new
914    /// reader is dropped and the caller gets [`DbError::ReadPoolClosed`].
915    fn open_checked_out(&self) -> Result<PooledReadConnection> {
916        match self.inner.open_reader() {
917            Ok(conn) => {
918                let mut state = lock_recover(&self.inner.state);
919                if state.closed {
920                    drop(conn);
921                    state.total -= 1;
922                    self.inner.drained.notify_all();
923                    return Err(DbError::ReadPoolClosed);
924                }
925                Ok(PooledReadConnection {
926                    conn: Some(conn),
927                    pool: Arc::downgrade(&self.inner),
928                })
929            }
930            Err(error) => {
931                let mut state = lock_recover(&self.inner.state);
932                state.total -= 1;
933                self.inner.available.notify_one();
934                self.inner.drained.notify_all();
935                Err(error)
936            }
937        }
938    }
939
940    /// Stops handing out readers and drops the idle ones now. A reader still checked out closes
941    /// when it is returned. Calling it again does nothing.
942    pub fn close(&self) {
943        let idle = {
944            let mut state = lock_recover(&self.inner.state);
945            if state.closed {
946                return;
947            }
948            state.closed = true;
949            self.inner.available.notify_all();
950            std::mem::take(&mut state.idle)
951        };
952        let closed = idle.len();
953        // Files close before the count drops, so `close_and_wait` sees them released.
954        drop(idle);
955        let mut state = lock_recover(&self.inner.state);
956        state.total -= closed;
957        self.inner.drained.notify_all();
958    }
959
960    /// [`Self::close`], then waits up to `timeout` for every checked-out reader to be returned.
961    /// `Ok(())` means every connection is closed and its database file released. Otherwise it
962    /// returns [`DbError::ReadPoolCloseTimeout`] with the number still out. Do not call it while
963    /// holding a reader from this pool: that reader cannot come back, so the call times out. A
964    /// timeout too large for the clock waits in bounded slices until the readers return.
965    pub fn close_and_wait(&self, timeout: Duration) -> Result<()> {
966        self.close();
967        let deadline = Instant::now().checked_add(timeout);
968        let mut state = lock_recover(&self.inner.state);
969        loop {
970            if state.total == 0 {
971                return Ok(());
972            }
973            let remaining = match deadline {
974                Some(at) => at.saturating_duration_since(Instant::now()),
975                None => Duration::from_secs(60),
976            };
977            if remaining.is_zero() {
978                return Err(DbError::ReadPoolCloseTimeout {
979                    outstanding: state.total,
980                    timeout,
981                });
982            }
983            let (next, _) = match self.inner.drained.wait_timeout(state, remaining) {
984                Ok(pair) => pair,
985                Err(poisoned) => poisoned.into_inner(),
986            };
987            state = next;
988        }
989    }
990
991    pub fn capacity(&self) -> usize {
992        self.inner.cap
993    }
994
995    pub fn idle_len(&self) -> usize {
996        lock_recover(&self.inner.state).idle.len()
997    }
998
999    pub fn total_len(&self) -> usize {
1000        lock_recover(&self.inner.state).total
1001    }
1002}
1003
1004fn lock_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
1005    mutex
1006        .lock()
1007        .unwrap_or_else(|poisoned| poisoned.into_inner())
1008}
1009
1010#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1011pub struct CheckpointStatus {
1012    pub busy: i64,
1013    pub log_frames: i64,
1014    pub checkpointed_frames: i64,
1015}
1016
1017pub fn checkpoint_truncate(conn: &Connection) -> Result<CheckpointStatus> {
1018    checkpoint(conn, "PRAGMA wal_checkpoint(TRUNCATE)")
1019}
1020
1021pub fn checkpoint_if_wal_larger_than(
1022    conn: &Connection,
1023    db: &Path,
1024    bytes: u64,
1025) -> Result<Option<CheckpointStatus>> {
1026    let wal = wal_path(db)?;
1027    let size = match std::fs::metadata(wal) {
1028        Ok(metadata) => metadata.len(),
1029        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
1030        Err(_) => return Ok(None),
1031    };
1032    if size <= bytes {
1033        return Ok(None);
1034    }
1035    checkpoint(conn, "PRAGMA wal_checkpoint(PASSIVE)").map(Some)
1036}
1037
1038fn checkpoint(conn: &Connection, pragma: &str) -> Result<CheckpointStatus> {
1039    conn.query_row(pragma, [], |row| {
1040        Ok(CheckpointStatus {
1041            busy: row.get(0)?,
1042            log_frames: row.get(1)?,
1043            checkpointed_frames: row.get(2)?,
1044        })
1045    })
1046    .map_err(Into::into)
1047}
1048
1049fn wal_path(db: &Path) -> Result<PathBuf> {
1050    let file_name = db.file_name().ok_or(DbError::InvalidDatabasePath)?;
1051    let mut wal_name = OsString::from(file_name);
1052    wal_name.push("-wal");
1053    Ok(db.with_file_name(wal_name))
1054}
1055
1056pub fn optimize_on_open(conn: &Connection) -> Result<()> {
1057    conn.execute_batch("PRAGMA optimize=0x10002;")?;
1058    Ok(())
1059}
1060
1061pub fn optimize_boundary(conn: &Connection) -> Result<()> {
1062    conn.execute_batch("PRAGMA optimize;")?;
1063    Ok(())
1064}
rightkit_db/lib.rs

rightkit_db/
lib.rs
