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

rightkit_db/
lib.rs
