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

rightkit_db/
lib.rs
