Skip to main content

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-libtomcrypt", feature = "sqlcipher-system"))]
5compile_error!("rightkit-db features `sqlcipher-libtomcrypt` and `sqlcipher-system` are mutually exclusive; enable exactly one");
6
7use rusqlite::config::DbConfig;
8use rusqlite::{Connection, OpenFlags, Transaction, TransactionBehavior};
9use std::ffi::OsString;
10use std::ops::Deref;
11use std::path::{Path, PathBuf};
12use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
13use std::time::{Duration, Instant};
14/// The zeroizing wrapper that the SQLCipher key fields and key candidates use, re-exported so
15/// callers build keys without declaring their own `zeroize` dependency.
16pub use zeroize::Zeroizing;
17
18mod backup;
19mod checkpoint_job;
20#[allow(unsafe_code)]
21mod host;
22pub use host::per_host_path;
23/// The exact `rusqlite` this crate links (`links = "sqlite3"` allows one per build). Apps name
24/// `Connection`, `params!`, `Row`, `OptionalExtension`, `TransactionBehavior` through this path
25/// instead of declaring their own `rusqlite` dependency.
26pub use rusqlite;
27mod migrate;
28mod recover;
29mod restore;
30mod unreadable;
31
32#[cfg(all(feature = "encrypted-lifecycle", not(feature = "cipher")))]
33compile_error!("rightkit-db feature `encrypted-lifecycle` needs a SQLCipher backend; also enable `sqlcipher-libtomcrypt` or `sqlcipher-system`");
34
35/// The encrypted-store lifecycle: key selection, rotation, recovery, restore and relocation behind
36/// one cross-process guard. Needs `encrypted-lifecycle` and a SQLCipher backend.
37#[cfg(all(feature = "encrypted-lifecycle", feature = "cipher"))]
38pub mod encrypted_lifecycle;
39
40pub use backup::{
41    online_backup_to, remove_database_files, replace_database, replace_database_locked,
42    restore_from, store_lock_path, BackupProgress, StoreLock,
43};
44pub use checkpoint_job::{
45    WalCheckpointJob, WalCheckpointMode, WalCheckpointPolicy, WalCheckpointRun,
46};
47pub use migrate::{
48    add_column_if_missing, column_exists, migrate, migration_version, pending_marker,
49    pending_marker_started_at, rollback_to, rollback_to_with_storage, stored_version_is_newer,
50    user_version, write_pending_marker, Degraded, LegacyVersionTable, MigrateOptions, Migration,
51    MigrationReport, OpenOrDegrade, PendingMarker, Step, VersionStorage,
52};
53#[cfg(feature = "cipher")]
54pub use recover::rekey;
55pub use recover::{
56    quarantine_database, quarantine_database_opts, quarantine_database_with, MoveMode, OpenOutcome,
57    QuarantineOptions, QuarantineStamp, RecoveryPolicy, SidecarPolicy,
58};
59
60pub use restore::{
61    recover_interrupted_restore, recover_interrupted_restore_locked, restore_database,
62    restore_database_locked, InterruptedRestore, RestoreOptions, RestoreReceipt, RestoreVerify,
63};
64
65pub use unreadable::{
66    classify_open_failure, unreadable_counter_path, RecoveryAction, StorageProblem,
67    UnreadableCause, UnreadablePolicy, UnreadableRecovery,
68};
69
70const BUSY_TIMEOUT_MS: u64 = 5_000;
71const JOURNAL_SIZE_LIMIT: i64 = 67_108_864;
72
73pub type Result<T> = std::result::Result<T, DbError>;
74
75#[derive(Debug, thiserror::Error)]
76#[non_exhaustive]
77pub enum DbError {
78    #[error("SQLite error: {0}")]
79    Sqlite(#[from] rusqlite::Error),
80    #[error("a SQLCipher key requires the rightkit-db `sqlcipher-libtomcrypt` or `sqlcipher-system` feature")]
81    SqlcipherFeatureRequired,
82    #[error("SQLCipher key contains characters that are unsafe for PRAGMA interpolation")]
83    InvalidSqlcipherKey,
84    /// An open-time `quick_check` or `integrity_check` failed. Open-time recovery treats it as an
85    /// unreadable database (see `DbOptions::open_writer_recovering`).
86    #[error("database integrity check failed: {0}")]
87    Integrity(String),
88    #[error("quarantine suffix {0:?} must be non-empty and contain no path separator or NUL")]
89    InvalidQuarantineSuffix(String),
90    #[error("read-pool capacity must be greater than zero")]
91    InvalidPoolCapacity,
92    #[error("reader cache size must be between 1 and 2147483647 KiB, got {0}")]
93    InvalidReaderCacheSize(u32),
94    /// Checkout after [`ReadPool::close`] or [`ReadPool::close_and_wait`].
95    #[error("read pool is closed")]
96    ReadPoolClosed,
97    /// [`ReadPool::close_and_wait`] timed out. `outstanding` counts readers still checked out or
98    /// opening.
99    #[error("read pool close timed out after {timeout:?} with {outstanding} reader(s) still out")]
100    ReadPoolCloseTimeout {
101        outstanding: usize,
102        timeout: Duration,
103    },
104    /// A reader did not come up read-only: `PRAGMA query_only` did not stay on, or a read-only
105    /// handle reported a writable main database.
106    #[error("reader connection is not read-only: {0}")]
107    ReadOnlyNotEffective(String),
108    #[error("database path has no file name")]
109    InvalidDatabasePath,
110    /// The store path is a symbolic link (a dangling one included) and was refused before SQLite
111    /// could create or open its target. Carries the path as given.
112    #[error("store path {0} is a symbolic link; refusing to open it")]
113    SymlinkedStorePath(String),
114    #[error("database is at schema v{current}, newer than this build understands (v{max}) and not forward compatible")]
115    NewerDatabase { current: i64, max: i64 },
116    #[error("invalid migration list: {0}")]
117    InvalidMigrations(String),
118    #[error("migration v{version} ({name}) failed and was rolled back: {source}")]
119    MigrationFailed {
120        version: i64,
121        name: &'static str,
122        #[source]
123        source: rusqlite::Error,
124    },
125    #[error("cannot roll back: {0}")]
126    Rollback(String),
127    #[error("database backup failed: {0}")]
128    Backup(String),
129    /// [`restore_database`] failed. `previous` is where the previous store is kept when it is not
130    /// at the live path (never deleted); `failed_copy` is where the restored copy was set aside.
131    /// `rolled_back` is true when the live path holds the previous store again (or was never
132    /// touched); false means the rollback was incomplete, a restore journal remains, and
133    /// [`recover_interrupted_restore`] finishes it.
134    #[error("database restore failed: {message}")]
135    RestoreFailed {
136        message: String,
137        previous: Option<PathBuf>,
138        failed_copy: Option<PathBuf>,
139        rolled_back: bool,
140    },
141    #[error("no candidate key opened the database")]
142    NoKeyOpened,
143    #[error(transparent)]
144    Fs(#[from] rightkit_fs::FsError),
145    /// The database file could not be read (permission or I/O failure) or its storage is
146    /// unavailable. Not corruption: the file was not touched unless `quarantine_after` was
147    /// reached, in which case recovery succeeds instead of returning this. See
148    /// [`UnreadablePolicy`].
149    #[error(
150        "database {} cannot be read: {cause} ({consecutive_failures} consecutive failed open(s)): {source}",
151        .path.display()
152    )]
153    Unreadable {
154        path: PathBuf,
155        cause: UnreadableCause,
156        /// Consecutive failed opens recorded so far (0 under [`UnreadablePolicy::Fail`] and for
157        /// storage problems, which are never counted).
158        consecutive_failures: u32,
159        /// The count at which [`UnreadablePolicy::QuarantineAfter`] would set the file aside.
160        quarantine_after: Option<u32>,
161        #[source]
162        source: Box<DbError>,
163    },
164}
165
166#[derive(Clone, Copy, Debug, Eq, PartialEq)]
167pub enum CacheProfile {
168    Default,
169    LargeWorkingSet,
170}
171
172#[derive(Clone, Copy, Debug, Eq, PartialEq)]
173pub enum DurabilityProfile {
174    Full,
175    Normal,
176}
177
178/// How the journal mode is chosen. WAL needs working shared-memory locking, which network
179/// filesystems often lack and which SQLite will silently downgrade; the default therefore
180/// *verifies* the mode that took effect and falls back to a rollback journal.
181#[derive(Clone, Copy, Debug, Eq, PartialEq)]
182#[non_exhaustive]
183pub enum JournalSelect {
184    /// WAL when the filesystem honors it and is not a detected network mount; else a
185    /// rollback journal (TRUNCATE with [`DbOptions::per_host_fallback`], else DELETE).
186    WalWithFallback,
187    /// WAL or fail (no fallback).
188    WalStrict,
189    /// Rollback journal, deleted after each transaction.
190    Delete,
191    /// Rollback journal, truncated after each transaction.
192    Truncate,
193}
194
195impl JournalSelect {
196    fn is_rollback(self) -> bool {
197        matches!(self, JournalSelect::Delete | JournalSelect::Truncate)
198    }
199}
200
201/// Environment kill-switch (`wal` | `delete` | `truncate`) for a misclassified mount in the
202/// field. An app's own variable ([`DbOptions::journal_env`]) is checked first.
203pub const JOURNAL_ENV: &str = "RIGHTKIT_SQLITE_JOURNAL_MODE";
204
205/// Post-key, pre-pragma per-connection initializer (register functions, collations, ...).
206/// Open-time integrity check, selected by [`DbOptions::integrity`].
207#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
208pub enum IntegrityCheck {
209    /// No check at open (default).
210    #[default]
211    Off,
212    /// `PRAGMA quick_check`: structural page checks, without the index-to-table cross-check. It
213    /// costs far less than `Full` on large databases.
214    Quick,
215    /// `PRAGMA integrity_check`, or `PRAGMA cipher_integrity_check` for a keyed database.
216    Full,
217}
218
219pub type ConnInit = Arc<dyn Fn(&Connection) -> rusqlite::Result<()> + Send + Sync>;
220
221#[derive(Clone)]
222pub struct DbOptions {
223    pub path: PathBuf,
224    /// SQLite busy wait, installed before key verification or any bootstrap statement.
225    /// Default: five seconds.
226    pub busy_timeout: Duration,
227    /// SQLCipher passphrase-style key (base64-ish charset). See also [`DbOptions::raw_key`].
228    pub key: Option<Zeroizing<String>>,
229    /// SQLCipher raw 256-bit key, applied as `x'<hex>'` (no KDF). Takes precedence over `key`.
230    pub raw_key: Option<Zeroizing<[u8; 32]>>,
231    pub profile: CacheProfile,
232    pub durability: DurabilityProfile,
233    pub foreign_keys: bool,
234    /// Legacy switch: `true` runs a full check on every open. Ignored when `integrity` is not `Off`.
235    pub integrity_check_on_open: bool,
236    /// Open-time integrity check (default `Off`). A failed check is [`DbError::Integrity`], which
237    /// `open_writer_recovering` treats as unreadable.
238    pub integrity: IntegrityCheck,
239    pub journal: JournalSelect,
240    /// Narrow the database and its sidecars to owner-only (0600) on unix. Default on. Set it to
241    /// `false` when sibling processes must read the file (for example a helper running as another
242    /// user); that is the intended switch, not a bug.
243    pub owner_only: bool,
244    pub init: Option<ConnInit>,
245    /// The app's own kill-switch variable (e.g. `CODERIGHT_SQLITE_JOURNAL_MODE`), checked
246    /// before [`JOURNAL_ENV`].
247    pub journal_env: Option<String>,
248    /// When a rollback journal is in force (network mount, kill switch, or WAL refused),
249    /// open the per-host sibling [`per_host_path`] instead of the requested file, so hosts
250    /// sharing a network directory never share one rollback-journaled database. The
251    /// requested file is not created. Default off.
252    pub per_host_fallback: bool,
253    /// Sets `PRAGMA secure_delete` on every connection at open: `Some(true)` overwrites freed
254    /// pages, `Some(false)` does not, `None` (default) leaves SQLite's compile-time default.
255    pub secure_delete: Option<bool>,
256    /// Run `PRAGMA optimize` on each writer connection at open (default `true`). Set it to `false`
257    /// to leave planner statistics alone at open, for example when the file is opened only to
258    /// inspect it. Readers never run it, whatever this is set to.
259    pub optimize_on_open: bool,
260}
261
262impl DbOptions {
263    pub fn new(path: impl Into<PathBuf>) -> Self {
264        Self {
265            path: path.into(),
266            busy_timeout: Duration::from_millis(BUSY_TIMEOUT_MS),
267            key: None,
268            raw_key: None,
269            profile: CacheProfile::Default,
270            durability: DurabilityProfile::Normal,
271            foreign_keys: false,
272            integrity_check_on_open: false,
273            integrity: IntegrityCheck::Off,
274            journal: JournalSelect::WalWithFallback,
275            owner_only: true,
276            init: None,
277            journal_env: None,
278            per_host_fallback: false,
279            secure_delete: None,
280            optimize_on_open: true,
281        }
282    }
283
284    pub fn with_journal_env(mut self, name: impl Into<String>) -> Self {
285        self.journal_env = Some(name.into());
286        self
287    }
288
289    pub fn with_per_host_fallback(mut self, on: bool) -> Self {
290        self.per_host_fallback = on;
291        self
292    }
293
294    /// Turn `PRAGMA optimize` off at writer open (see [`DbOptions::optimize_on_open`]).
295    pub fn without_optimize_on_open(mut self) -> Self {
296        self.optimize_on_open = false;
297        self
298    }
299
300    /// The file connections actually open: the requested path, or its per-host sibling when
301    /// [`Self::per_host_fallback`] applies (a rollback journal is in force, or a writer
302    /// already fell back there).
303    pub fn effective_path(&self) -> PathBuf {
304        if !self.per_host_fallback {
305            return self.path.clone();
306        }
307        let sibling = per_host_path(&self.path);
308        if effective_journal(self).is_rollback() || (sibling.exists() && !self.path.exists()) {
309            sibling
310        } else {
311            self.path.clone()
312        }
313    }
314
315    pub fn with_passphrase(mut self, key: impl Into<String>) -> Self {
316        self.key = Some(Zeroizing::new(key.into()));
317        self.raw_key = None;
318        self
319    }
320
321    pub fn with_raw_key(mut self, key: [u8; 32]) -> Self {
322        self.raw_key = Some(Zeroizing::new(key));
323        self.key = None;
324        self
325    }
326
327    pub fn is_keyed(&self) -> bool {
328        self.key.is_some() || self.raw_key.is_some()
329    }
330
331    pub fn open_writer(&self) -> Result<Connection> {
332        self.open(ConnectionKind::Writer)
333    }
334
335    pub fn open_reader(&self) -> Result<Connection> {
336        self.open(ConnectionKind::Reader { cache_kib: None })
337    }
338
339    /// Open exactly `path` read-only, without creating it or running mutating bootstrap.
340    /// URI parameters are passed to SQLite unchanged. Journal fallback, permissions,
341    /// `init`, optimization, and migrations are skipped; key and timeout still apply.
342    pub fn open_read_only(&self) -> Result<Connection> {
343        self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_ONLY)
344    }
345
346    /// Open exactly `path` read-write without CREATE or mutating bootstrap.
347    /// Uses the same minimal setup as [`Self::open_read_only`].
348    pub fn open_existing_writer(&self) -> Result<Connection> {
349        self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_WRITE)
350    }
351
352    /// Open a private, ephemeral database. `path`, filesystem permissions, journal
353    /// selection/fallback, and mmap are ignored; other bootstrap options still apply.
354    pub fn open_in_memory(&self) -> Result<Connection> {
355        let key = self.key_pragma_value()?;
356        let conn = Connection::open_in_memory()?;
357        self.initialize(&conn, key.as_ref().map(|k| k.as_str()))?;
358        configure_connection(&conn, self, ConnectionKind::Memory, None)?;
359        self.check_on_open(&conn)?;
360        Ok(conn)
361    }
362
363    fn open_existing_at(&self, path: &Path, access: OpenFlags) -> Result<Connection> {
364        let key = self.key_pragma_value()?;
365        let writable = access.contains(OpenFlags::SQLITE_OPEN_READ_WRITE);
366        let _pin = pin_store_path(path, writable, false)?;
367        let conn = open_store(
368            path,
369            access | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX,
370        )?;
371        conn.busy_timeout(self.busy_timeout)?;
372        apply_key_first(&conn, key.as_ref().map(|k| k.as_str()))?;
373        verify_key(&conn)?;
374        self.check_on_open(&conn)?;
375        Ok(conn)
376    }
377
378    fn initialize(&self, conn: &Connection, key: Option<&str>) -> Result<()> {
379        conn.busy_timeout(self.busy_timeout)?;
380        apply_key_first(conn, key)?;
381        verify_key(conn)?;
382        if let Some(init) = &self.init {
383            init(conn)?;
384        }
385        Ok(())
386    }
387
388    /// A read pool with [`ReadPoolOptions::new`]: read-write file handles, with `query_only` set
389    /// and verified on every reader. Same as `open_read_pool_with(cap, ReadPoolOptions::new())`.
390    pub fn open_read_pool(&self, cap: usize) -> Result<ReadPool> {
391        ReadPool::open(self.clone(), cap, ReadPoolOptions::new())
392    }
393
394    /// A read pool configured by [`ReadPoolOptions`]. A read-only pool needs the database to exist.
395    pub fn open_read_pool_with(&self, cap: usize, pool: ReadPoolOptions) -> Result<ReadPool> {
396        ReadPool::open(self.clone(), cap, pool)
397    }
398
399    fn open_pool_reader(&self, pool: ReadPoolOptions) -> Result<Connection> {
400        if pool.read_only {
401            self.open_pool_read_only(pool.cache_kib)
402        } else {
403            self.open(ConnectionKind::Reader {
404                cache_kib: pool.cache_kib,
405            })
406        }
407    }
408
409    /// A pool reader opened with `SQLITE_OPEN_READ_ONLY`, so the file refuses writes even after
410    /// `query_only` is turned off. It never creates the file, changes the journal mode, narrows
411    /// permissions, or runs `optimize`; the writer owns those. `init` runs on each reader.
412    fn open_pool_read_only(&self, cache_kib: Option<u32>) -> Result<Connection> {
413        let path = self.effective_path();
414        let key = self.key_pragma_value()?;
415        let _pin = pin_store_path(&path, false, false)?;
416        let conn = open_store(
417            &path,
418            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
419        )?;
420        conn.busy_timeout(self.busy_timeout)?;
421        apply_key_first(&conn, key.as_ref().map(|k| k.as_str()))?;
422        verify_key(&conn)?;
423        conn.pragma_update(None, "temp_store", "MEMORY")?;
424        if self.foreign_keys {
425            conn.pragma_update(None, "foreign_keys", true)?;
426        }
427        apply_cache_settings(&conn, self, ConnectionKind::Reader { cache_kib })?;
428        // Per-connection setup (`init`) runs on every reader, read-only ones included, as it does
429        // on the other connection kinds. An `init` that writes fails here with the SQLite error.
430        if let Some(init) = &self.init {
431            init(&conn)?;
432        }
433        enforce_query_only(&conn)?;
434        if !conn.is_readonly("main")? {
435            return Err(DbError::ReadOnlyNotEffective(
436                "SQLITE_OPEN_READ_ONLY handle reports a writable main database".into(),
437            ));
438        }
439        self.check_on_open(&conn)?;
440        Ok(conn)
441    }
442
443    fn open(&self, kind: ConnectionKind) -> Result<Connection> {
444        let path = self.effective_path();
445        // The per-host sibling is always rollback-journaled, whoever opens it.
446        let forced = (path != self.path).then(|| match effective_journal(self) {
447            mode if mode.is_rollback() => mode,
448            _ => fallback_mode(self),
449        });
450        let (conn, journal) = self.open_at(&path, kind, forced)?;
451        // WAL refused at runtime: with per-host fallback, the rollback journal belongs on the
452        // per-host sibling. The main file this open touched stays where it is, with every
453        // `-wal`/`-shm`/`-journal` beside it. Deleting it is never safe: file identity proves
454        // origin, not exclusive lifetime, and another connection may have committed
455        // rollback-journal data into that same file (and closed) before this point, so any
456        // check-then-remove can destroy a committed write. A created-empty main is harmless
457        // (it never holds data this open wrote, and a later open falls back again).
458        if journal == Opened::FellBack && self.per_host_fallback && path == self.path {
459            drop(conn);
460            let sibling = per_host_path(&self.path);
461            return Ok(self
462                .open_at(&sibling, kind, Some(JournalSelect::Truncate))?
463                .0);
464        }
465        Ok(conn)
466    }
467
468    fn open_at(
469        &self,
470        path: &Path,
471        kind: ConnectionKind,
472        forced: Option<JournalSelect>,
473    ) -> Result<(Connection, Opened)> {
474        let key = self.key_pragma_value()?;
475        let flags = OpenFlags::SQLITE_OPEN_READ_WRITE
476            | OpenFlags::SQLITE_OPEN_CREATE
477            | OpenFlags::SQLITE_OPEN_NO_MUTEX;
478        // The store file is refused as a link at the open itself, never by a check before it:
479        // `open_store` passes SQLITE_OPEN_NOFOLLOW, and on Windows `pin_store_path` holds the file
480        // so the path cannot be swapped in between. The pin lives until the connection is open.
481        let _pin = pin_store_path(path, true, true)?;
482        let conn = open_store(path, flags)?;
483        // A failed open drops `conn`. A default close checkpoints and then deletes the `-wal` and
484        // `-shm`, which destroys the evidence quarantine promises to keep (and rewrites a valid but
485        // corrupt main file). Suppress that until the open has fully succeeded. This is a plain
486        // connection setting, valid before keying; it reads nothing from the file.
487        conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, true)?;
488        // Narrow the main file before any write so WAL/SHM sidecars inherit owner-only.
489        if self.owner_only {
490            rightkit_fs::restrict_owner_only(path)?;
491        }
492        self.initialize(&conn, key.as_ref().map(|k| k.as_str()))?;
493        if let Some(on) = self.secure_delete {
494            conn.pragma_update(None, "secure_delete", if on { "ON" } else { "OFF" })?;
495        }
496        let journal = configure_connection(&conn, self, kind, forced)?;
497        self.check_on_open(&conn)?;
498        if self.owner_only {
499            harden_sidecars(path)?;
500        }
501        // The open succeeded: restore the normal close behaviour (checkpoint and remove the WAL).
502        conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, false)?;
503        Ok((conn, journal))
504    }
505
506    /// Run the open-time check selected by `integrity`; `integrity_check_on_open` alone means `Full`.
507    fn check_on_open(&self, conn: &Connection) -> Result<()> {
508        let mode = match self.integrity {
509            IntegrityCheck::Off if self.integrity_check_on_open => IntegrityCheck::Full,
510            mode => mode,
511        };
512        match (mode, self.is_keyed()) {
513            (IntegrityCheck::Off, _) => Ok(()),
514            (IntegrityCheck::Quick, _) => run_integrity_pragma(conn, "PRAGMA quick_check"),
515            (IntegrityCheck::Full, true) => {
516                run_integrity_pragma(conn, "PRAGMA cipher_integrity_check")
517            }
518            (IntegrityCheck::Full, false) => run_integrity_pragma(conn, "PRAGMA integrity_check"),
519        }
520    }
521
522    /// The `PRAGMA key` right-hand side (already quoted), validated against injection.
523    fn key_pragma_value(&self) -> Result<Option<Zeroizing<String>>> {
524        if let Some(raw) = &self.raw_key {
525            #[cfg(not(feature = "cipher"))]
526            {
527                let _ = raw;
528                return Err(DbError::SqlcipherFeatureRequired);
529            }
530            #[cfg(feature = "cipher")]
531            {
532                let hex: String = raw.iter().map(|b| format!("{b:02x}")).collect();
533                return Ok(Some(Zeroizing::new(format!("\"x'{hex}'\""))));
534            }
535        }
536        let key = self.key.as_ref().map(|key| key.as_str());
537        validate_key(key)?;
538        Ok(key.map(|k| Zeroizing::new(format!("'{k}'"))))
539    }
540}
541
542/// Default-options read-only open; never creates a missing database.
543pub fn open_read_only(path: impl AsRef<Path>) -> Result<Connection> {
544    DbOptions::new(path.as_ref()).open_read_only()
545}
546
547/// Default-options read-write open; never creates a missing database.
548pub fn open_existing_writer(path: impl AsRef<Path>) -> Result<Connection> {
549    DbOptions::new(path.as_ref()).open_existing_writer()
550}
551
552/// Default-options private in-memory database.
553pub fn open_in_memory() -> Result<Connection> {
554    DbOptions::new(":memory:").open_in_memory()
555}
556
557/// Open the store file through SQLite with `SQLITE_OPEN_NOFOLLOW`, so a link at the store file's own
558/// name is refused by SQLite (dangling links included: neither the link nor its target is created or
559/// opened). SQLite's `NOFOLLOW` check is by path resolution, and it counts a link in any directory on
560/// the path too (macOS `/var` is one), so on Unix the parent directory is canonicalized first and
561/// only the file name is left to the check. SQLite's Unix VFS also opens the final component with
562/// `O_NOFOLLOW`, so a link swapped in after SQLite's check still fails the open itself. A failed open
563/// whose path is a link is [`DbError::SymlinkedStorePath`]; any other failure keeps SQLite's error.
564fn open_store(path: &Path, flags: OpenFlags) -> Result<Connection> {
565    let target = store_open_target(path);
566    Connection::open_with_flags(&target, flags | OpenFlags::SQLITE_OPEN_NOFOLLOW).map_err(|error| {
567        if std::fs::symlink_metadata(path).is_ok_and(|meta| meta.file_type().is_symlink()) {
568            DbError::SymlinkedStorePath(path.display().to_string())
569        } else {
570            DbError::Sqlite(error)
571        }
572    })
573}
574
575/// The path handed to SQLite: on Unix the canonical parent joined with the file name, so that
576/// `NOFOLLOW` judges only the store file; elsewhere, or when the parent cannot be resolved (SQLite
577/// then reports the missing directory itself), the path as given.
578#[cfg(unix)]
579fn store_open_target(path: &Path) -> PathBuf {
580    match (path.parent(), path.file_name()) {
581        (Some(dir), Some(name)) if !dir.as_os_str().is_empty() => std::fs::canonicalize(dir)
582            .map(|dir| dir.join(name))
583            .unwrap_or_else(|_| path.to_path_buf()),
584        _ => path.to_path_buf(),
585    }
586}
587
588#[cfg(not(unix))]
589fn store_open_target(path: &Path) -> PathBuf {
590    path.to_path_buf()
591}
592
593/// Windows has no `O_NOFOLLOW`: SQLite's Windows VFS opens a link's target. The store file is
594/// therefore opened here as the reparse point itself, refused if it is a link, and the handle is
595/// held (without delete sharing, so the path cannot be renamed or deleted and replaced) until the
596/// connection is open. Unix needs no pin: [`open_store`] and SQLite's own `O_NOFOLLOW` do the work.
597/// `write` and `create` mirror the SQLite open; a writable open that the OS refuses falls back to a
598/// read-only pin, as SQLite does for a read-only file.
599#[cfg(windows)]
600fn pin_store_path(path: &Path, write: bool, create: bool) -> Result<Option<std::fs::File>> {
601    use std::os::windows::fs::OpenOptionsExt;
602    /// Open the reparse point itself instead of following it.
603    const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
604    const FILE_SHARE_READ: u32 = 0x1;
605    const FILE_SHARE_WRITE: u32 = 0x2;
606    let open = |write: bool, create: bool| {
607        std::fs::OpenOptions::new()
608            .read(true)
609            .write(write)
610            .create(create)
611            .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE)
612            .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT)
613            .open(path)
614    };
615    let file = match open(write, create) {
616        Ok(file) => file,
617        Err(error) if write && error.kind() == std::io::ErrorKind::PermissionDenied => {
618            match open(false, false) {
619                Ok(file) => file,
620                Err(error) => return pin_error(path, error),
621            }
622        }
623        Err(error) => return pin_error(path, error),
624    };
625    let meta = file
626        .metadata()
627        .map_err(|e| rightkit_fs::FsError::io(path, e))?;
628    if meta.file_type().is_symlink() {
629        return Err(DbError::SymlinkedStorePath(path.display().to_string()));
630    }
631    Ok(Some(file))
632}
633
634/// A missing file or directory is left to SQLite, which reports it (or creates the file) as it always
635/// has; any other failure to open the store file is an error.
636#[cfg(windows)]
637fn pin_error(path: &Path, error: std::io::Error) -> Result<Option<std::fs::File>> {
638    if error.kind() == std::io::ErrorKind::NotFound {
639        Ok(None)
640    } else {
641        Err(rightkit_fs::FsError::io(path, error).into())
642    }
643}
644
645#[cfg(not(windows))]
646fn pin_store_path(_path: &Path, _write: bool, _create: bool) -> Result<Option<std::fs::File>> {
647    Ok(None)
648}
649
650fn harden_sidecars(db: &Path) -> Result<()> {
651    for suffix in ["-wal", "-shm"] {
652        let mut name = OsString::from(db.file_name().ok_or(DbError::InvalidDatabasePath)?);
653        name.push(suffix);
654        rightkit_fs::restrict_owner_only(&db.with_file_name(name))?;
655    }
656    Ok(())
657}
658
659#[derive(Clone, Copy)]
660enum ConnectionKind {
661    Writer,
662    /// `cache_kib` overrides the reader cache size; see [`ReadPoolOptions::with_reader_cache_kib`].
663    Reader {
664        cache_kib: Option<u32>,
665    },
666    Memory,
667}
668
669impl ConnectionKind {
670    fn reader_cache_kib(self) -> Option<u32> {
671        match self {
672            ConnectionKind::Reader { cache_kib } => cache_kib,
673            ConnectionKind::Writer | ConnectionKind::Memory => None,
674        }
675    }
676}
677
678/// What journal setup did.
679#[derive(Clone, Copy, Debug, Eq, PartialEq)]
680enum Opened {
681    AsRequested,
682    /// WAL was wanted but not honored; a rollback journal is in force.
683    FellBack,
684}
685
686fn configure_connection(
687    conn: &Connection,
688    options: &DbOptions,
689    kind: ConnectionKind,
690    forced: Option<JournalSelect>,
691) -> Result<Opened> {
692    let opened = if matches!(kind, ConnectionKind::Memory) {
693        Opened::AsRequested
694    } else {
695        configure_journal(conn, options, kind, forced)?
696    };
697    let synchronous = match options.durability {
698        DurabilityProfile::Full => "FULL",
699        DurabilityProfile::Normal => "NORMAL",
700    };
701    conn.pragma_update(None, "synchronous", synchronous)?;
702    conn.pragma_update(None, "journal_size_limit", JOURNAL_SIZE_LIMIT)?;
703    conn.pragma_update(None, "temp_store", "MEMORY")?;
704    if options.foreign_keys {
705        conn.pragma_update(None, "foreign_keys", true)?;
706    }
707    apply_cache_settings(conn, options, kind)?;
708    if matches!(kind, ConnectionKind::Reader { .. }) {
709        // Readers never run `PRAGMA optimize`: it can write statistics, and the writer runs it
710        // when it opens. `query_only` is set after every other reader setup statement.
711        enforce_query_only(conn)?;
712    } else if options.optimize_on_open {
713        optimize_on_open(conn)?;
714    }
715    Ok(opened)
716}
717
718/// Profile cache and mmap sizes, then the reader override, which wins over the profile on readers
719/// only. Writers and in-memory connections keep the profile size.
720fn apply_cache_settings(
721    conn: &Connection,
722    options: &DbOptions,
723    kind: ConnectionKind,
724) -> Result<()> {
725    if options.profile == CacheProfile::LargeWorkingSet {
726        conn.pragma_update(None, "cache_size", -65_536_i64)?;
727        if !matches!(kind, ConnectionKind::Memory) {
728            conn.pragma_update(None, "mmap_size", 268_435_456_i64)?;
729        }
730    }
731    if let Some(kib) = kind.reader_cache_kib() {
732        conn.pragma_update(None, "cache_size", -i64::from(kib))?;
733    }
734    Ok(())
735}
736
737/// Sets `PRAGMA query_only = ON` on a reader and reads it back, so a reader that did not come up
738/// query-only is never handed out.
739fn enforce_query_only(conn: &Connection) -> Result<()> {
740    conn.pragma_update(None, "query_only", true)?;
741    let on: i64 = conn.query_row("PRAGMA query_only", [], |row| row.get(0))?;
742    if on != 1 {
743        return Err(DbError::ReadOnlyNotEffective(
744            "PRAGMA query_only did not stay ON".into(),
745        ));
746    }
747    Ok(())
748}
749
750fn set_rollback(conn: &Connection, mode: JournalSelect) -> Result<()> {
751    let pragma = if mode == JournalSelect::Truncate {
752        "PRAGMA journal_mode=TRUNCATE"
753    } else {
754        "PRAGMA journal_mode=DELETE"
755    };
756    conn.query_row(pragma, [], |row| row.get::<_, String>(0))?;
757    Ok(())
758}
759
760fn configure_journal(
761    conn: &Connection,
762    options: &DbOptions,
763    kind: ConnectionKind,
764    forced: Option<JournalSelect>,
765) -> Result<Opened> {
766    let wanted = forced.unwrap_or_else(|| effective_journal(options));
767    if wanted.is_rollback() {
768        // Rollback modes are per connection: a reader left in the default DELETE mode would
769        // try to delete a TRUNCATE writer's open journal (IOERR_DELETE on Windows).
770        set_rollback(conn, wanted)?;
771        return Ok(Opened::AsRequested);
772    }
773    let honored = conn
774        .query_row("PRAGMA journal_mode=WAL", [], |row| row.get::<_, String>(0))
775        .map(|mode| mode.eq_ignore_ascii_case("wal"))
776        .unwrap_or(false)
777        // `journal_mode=WAL` can report success before the -shm/-wal sidecars are usable;
778        // one empty write transaction proves the sidecars work before anything depends on them.
779        && (matches!(kind, ConnectionKind::Reader { .. })
780            || conn.execute_batch("BEGIN IMMEDIATE; ROLLBACK;").is_ok());
781    if honored {
782        return Ok(Opened::AsRequested);
783    }
784    if wanted == JournalSelect::WalStrict {
785        return Err(DbError::Sqlite(rusqlite::Error::SqliteFailure(
786            rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR),
787            Some("WAL journal mode was not honored by this filesystem".into()),
788        )));
789    }
790    let _ = conn.execute_batch("ROLLBACK;");
791    set_rollback(conn, fallback_mode(options))?;
792    Ok(Opened::FellBack)
793}
794
795/// The rollback journal used when WAL is unavailable: TRUNCATE with per-host fallback
796/// (CodeRight's network mode), DELETE otherwise.
797fn fallback_mode(options: &DbOptions) -> JournalSelect {
798    if options.per_host_fallback {
799        JournalSelect::Truncate
800    } else {
801        JournalSelect::Delete
802    }
803}
804
805/// Kill switches (the app's own variable first, then [`JOURNAL_ENV`]); set-but-empty or
806/// unrecognized values are ignored.
807fn env_journal(options: &DbOptions) -> Option<JournalSelect> {
808    options
809        .journal_env
810        .iter()
811        .map(String::as_str)
812        .chain([JOURNAL_ENV])
813        .find_map(|name| {
814            match std::env::var(name)
815                .ok()?
816                .trim()
817                .to_ascii_lowercase()
818                .as_str()
819            {
820                "wal" => Some(JournalSelect::WalStrict),
821                "delete" => Some(JournalSelect::Delete),
822                "truncate" => Some(JournalSelect::Truncate),
823                _ => None,
824            }
825        })
826}
827
828fn effective_journal(options: &DbOptions) -> JournalSelect {
829    if let Some(mode) = env_journal(options) {
830        return mode;
831    }
832    if options.journal == JournalSelect::WalWithFallback
833        && rightkit_fs::is_network_filesystem(&options.path)
834    {
835        return fallback_mode(options);
836    }
837    options.journal
838}
839
840/// The journal mode SQLite is actually running (`wal`, `delete`, ...).
841pub fn journal_mode(conn: &Connection) -> Result<String> {
842    Ok(conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))?)
843}
844
845/// Start a transaction that reserves SQLite's single writer slot up front.
846/// This keeps read-then-write work out of the unretryable BUSY_SNAPSHOT path.
847pub fn write_txn(conn: &mut Connection) -> Result<Transaction<'_>> {
848    conn.transaction_with_behavior(TransactionBehavior::Immediate)
849        .map_err(Into::into)
850}
851
852/// [`write_txn`] for a caller that holds only `&Connection`, such as a closure-style helper.
853/// It runs `BEGIN IMMEDIATE` through `Transaction::new_unchecked`, so the reserved writer slot is
854/// the same. The borrow checker does not stop a nested transaction here: SQLite refuses the
855/// `BEGIN` instead. The transaction rolls back when dropped unless it is committed.
856pub fn write_txn_unchecked(conn: &Connection) -> Result<Transaction<'_>> {
857    Transaction::new_unchecked(conn, TransactionBehavior::Immediate).map_err(Into::into)
858}
859
860#[cfg(feature = "cipher")]
861fn apply_key_first(conn: &Connection, key: Option<&str>) -> Result<()> {
862    if let Some(key) = key {
863        let statement = Zeroizing::new(format!("PRAGMA key = {key};"));
864        conn.execute_batch(statement.as_str())?;
865    }
866    Ok(())
867}
868
869#[cfg(not(feature = "cipher"))]
870fn apply_key_first(_conn: &Connection, key: Option<&str>) -> Result<()> {
871    if key.is_some() {
872        return Err(DbError::SqlcipherFeatureRequired);
873    }
874    Ok(())
875}
876
877fn validate_key(key: Option<&str>) -> Result<()> {
878    #[cfg(not(feature = "cipher"))]
879    if key.is_some() {
880        return Err(DbError::SqlcipherFeatureRequired);
881    }
882    if let Some(key) = key {
883        let safe = !key.is_empty()
884            && key.len() <= 4_096
885            && key.bytes().all(|byte| {
886                byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'/' | b'=' | b'-' | b'_')
887            });
888        if !safe {
889            return Err(DbError::InvalidSqlcipherKey);
890        }
891    }
892    Ok(())
893}
894
895fn verify_key(conn: &Connection) -> Result<()> {
896    conn.query_row("SELECT count(*) FROM sqlite_master", [], |row| {
897        row.get::<_, i64>(0)
898    })?;
899    Ok(())
900}
901
902fn run_integrity_pragma(conn: &Connection, pragma: &str) -> Result<()> {
903    let mut statement = conn.prepare(pragma)?;
904    let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
905    let messages = rows.collect::<std::result::Result<Vec<_>, _>>()?;
906    if messages.is_empty()
907        || messages
908            .iter()
909            .all(|message| message.eq_ignore_ascii_case("ok"))
910    {
911        return Ok(());
912    }
913    Err(DbError::Integrity(messages.join("; ")))
914}
915
916/// Options for a [`ReadPool`]. The default ([`ReadPoolOptions::new`]) matches
917/// [`DbOptions::open_read_pool`]: read-write file handles, with `query_only` set and verified on
918/// every reader. Build with the `with_*` methods; the fields are private.
919#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
920pub struct ReadPoolOptions {
921    read_only: bool,
922    cache_kib: Option<u32>,
923}
924
925impl ReadPoolOptions {
926    pub fn new() -> Self {
927        Self::default()
928    }
929
930    /// Opens each reader with `SQLITE_OPEN_READ_ONLY`, so writes fail even if `query_only` is later
931    /// turned off. The database must already exist: the pool never creates it, changes its journal
932    /// mode, narrows its permissions, or runs `optimize`. Each reader still runs the `init` hook,
933    /// as on the other connections. A rollback journal left by a crash is recovered by a writer,
934    /// not by a read-only reader. Default off.
935    pub fn with_read_only(mut self, on: bool) -> Self {
936        self.read_only = on;
937        self
938    }
939
940    /// Page cache size for each reader connection, in KiB. It is applied as `PRAGMA cache_size`
941    /// set to `-kib`, so the size does not depend on the page size. Valid values are 1 to
942    /// 2147483647; the pool checks this when it opens. It overrides
943    /// [`CacheProfile::LargeWorkingSet`] on readers only. Writers and [`DbOptions::open_reader`]
944    /// keep the profile size. Default: the profile's size.
945    pub fn with_reader_cache_kib(mut self, kib: u32) -> Self {
946        self.cache_kib = Some(kib);
947        self
948    }
949}
950
951/// A bounded pool of reader connections. Clones share one pool.
952#[derive(Clone)]
953pub struct ReadPool {
954    inner: Arc<ReadPoolInner>,
955}
956
957/// Opens one reader connection. A pool built from [`DbOptions`] uses that database's reader
958/// setup; [`ReadPool::from_opener`] supplies its own.
959type ReaderOpener = Box<dyn Fn() -> Result<Connection> + Send + Sync>;
960
961struct ReadPoolInner {
962    open: ReaderOpener,
963    cap: usize,
964    state: Mutex<ReadPoolState>,
965    /// Wakes a `get` waiting for capacity, and every waiter when the pool closes.
966    available: Condvar,
967    /// Wakes `close_and_wait` when a connection is released after the pool closed.
968    drained: Condvar,
969}
970
971#[derive(Default)]
972struct ReadPoolState {
973    idle: Vec<Connection>,
974    /// Idle, checked-out, and still-opening connections.
975    total: usize,
976    closed: bool,
977}
978
979impl ReadPoolInner {
980    fn open_reader(&self) -> Result<Connection> {
981        (self.open)()
982    }
983}
984
985pub struct PooledReadConnection {
986    conn: Option<Connection>,
987    pool: Weak<ReadPoolInner>,
988    // Managed lifecycle checkout retains its exact generation/guard until connection return.
989    _lifecycle_lease: Option<Arc<dyn Send + Sync>>,
990}
991
992impl PooledReadConnection {
993    #[cfg(feature = "encrypted-lifecycle")]
994    pub(crate) fn with_lifecycle_lease(mut self, lease: Arc<dyn Send + Sync>) -> Self {
995        self._lifecycle_lease = Some(lease);
996        self
997    }
998}
999
1000impl Deref for PooledReadConnection {
1001    type Target = Connection;
1002
1003    fn deref(&self) -> &Self::Target {
1004        self.conn.as_ref().expect("pooled connection is present")
1005    }
1006}
1007
1008impl Drop for PooledReadConnection {
1009    fn drop(&mut self) {
1010        let Some(conn) = self.conn.take() else {
1011            return;
1012        };
1013        let Some(pool) = self.pool.upgrade() else {
1014            return;
1015        };
1016        let mut state = lock_recover(&pool.state);
1017        if state.closed {
1018            // Close the file before the count drops, so `close_and_wait` returns only once it is
1019            // released. Checked under the same lock as `close`, so no connection is parked idle
1020            // after the pool has closed.
1021            drop(conn);
1022            state.total -= 1;
1023            pool.drained.notify_all();
1024        } else {
1025            state.idle.push(conn);
1026            pool.available.notify_one();
1027        }
1028    }
1029}
1030
1031impl ReadPool {
1032    fn open(options: DbOptions, cap: usize, pool_options: ReadPoolOptions) -> Result<Self> {
1033        if cap == 0 {
1034            return Err(DbError::InvalidPoolCapacity);
1035        }
1036        if let Some(kib) = pool_options.cache_kib {
1037            if kib == 0 || kib > i32::MAX.unsigned_abs() {
1038                return Err(DbError::InvalidReaderCacheSize(kib));
1039            }
1040        }
1041        Self::with_opener(
1042            cap,
1043            Box::new(move || options.open_pool_reader(pool_options)),
1044        )
1045    }
1046
1047    /// A pool whose readers come from `open`, called once now to check it works and again for
1048    /// each new reader. The pool enforces `cap` and the closed and wait rules; `query_only` and
1049    /// other reader setup are the opener's job. Intended for tests that need a pool over a
1050    /// connection they build themselves. Use [`DbOptions::open_read_pool`] for a real database.
1051    pub fn from_opener(
1052        cap: usize,
1053        open: impl Fn() -> Result<Connection> + Send + Sync + 'static,
1054    ) -> Result<Self> {
1055        Self::with_opener(cap, Box::new(open))
1056    }
1057
1058    fn with_opener(cap: usize, open: ReaderOpener) -> Result<Self> {
1059        if cap == 0 {
1060            return Err(DbError::InvalidPoolCapacity);
1061        }
1062        let pool = Self {
1063            inner: Arc::new(ReadPoolInner {
1064                open,
1065                cap,
1066                state: Mutex::new(ReadPoolState::default()),
1067                available: Condvar::new(),
1068                drained: Condvar::new(),
1069            }),
1070        };
1071        let conn = pool.inner.open_reader()?;
1072        let mut state = lock_recover(&pool.inner.state);
1073        state.total = 1;
1074        state.idle.push(conn);
1075        drop(state);
1076        Ok(pool)
1077    }
1078
1079    /// Checks out an idle reader, opens a new one while under the cap, or waits for a return.
1080    /// Fails with [`DbError::ReadPoolClosed`] once the pool is closed, including while waiting.
1081    /// The wait has no bound; use [`Self::get_timeout`] to give up after a deadline.
1082    pub fn get(&self) -> Result<PooledReadConnection> {
1083        match self.checkout(None)? {
1084            Some(reader) => Ok(reader),
1085            // Unreachable: an unbounded wait ends only with a reader or an error.
1086            None => Err(DbError::ReadPoolClosed),
1087        }
1088    }
1089
1090    /// [`Self::get`] with a bounded wait. Returns `Ok(Some(reader))` as soon as a reader is idle
1091    /// or can be opened under the cap, and `Ok(None)` if none frees up within `timeout`. A zero
1092    /// `timeout` never waits: it returns an idle or newly opened reader, or `Ok(None)`. If the
1093    /// pool is closed, or closes while waiting, the result is [`DbError::ReadPoolClosed`], never
1094    /// `Ok(None)`. Opening a new reader is not bounded by `timeout`. A timeout too large for the
1095    /// clock waits in bounded slices until a reader frees up or the pool closes.
1096    pub fn get_timeout(&self, timeout: Duration) -> Result<Option<PooledReadConnection>> {
1097        self.checkout(Some(timeout))
1098    }
1099
1100    /// Shared checkout. `wait` is `None` for no deadline.
1101    fn checkout(&self, wait: Option<Duration>) -> Result<Option<PooledReadConnection>> {
1102        let started = Instant::now();
1103        let deadline = wait.and_then(|wait| started.checked_add(wait));
1104        let mut state = lock_recover(&self.inner.state);
1105        loop {
1106            if state.closed {
1107                return Err(DbError::ReadPoolClosed);
1108            }
1109            if let Some(conn) = state.idle.pop() {
1110                return Ok(Some(PooledReadConnection {
1111                    conn: Some(conn),
1112                    pool: Arc::downgrade(&self.inner),
1113                    _lifecycle_lease: None,
1114                }));
1115            }
1116            if state.total < self.inner.cap {
1117                state.total += 1;
1118                drop(state);
1119                return self.open_checked_out().map(Some);
1120            }
1121            state = match (wait, deadline) {
1122                (None, _) => match self.inner.available.wait(state) {
1123                    Ok(state) => state,
1124                    Err(poisoned) => poisoned.into_inner(),
1125                },
1126                (Some(_), Some(at)) => {
1127                    let remaining = at.saturating_duration_since(Instant::now());
1128                    if remaining.is_zero() {
1129                        return Ok(None);
1130                    }
1131                    match self.inner.available.wait_timeout(state, remaining) {
1132                        Ok((state, _)) => state,
1133                        Err(poisoned) => poisoned.into_inner().0,
1134                    }
1135                }
1136                (Some(_), None) => {
1137                    match self
1138                        .inner
1139                        .available
1140                        .wait_timeout(state, Duration::from_secs(60))
1141                    {
1142                        Ok((state, _)) => state,
1143                        Err(poisoned) => poisoned.into_inner().0,
1144                    }
1145                }
1146            };
1147        }
1148    }
1149
1150    /// Opens a reader for a `get` that reserved a slot. If the pool closed meanwhile, the new
1151    /// reader is dropped and the caller gets [`DbError::ReadPoolClosed`].
1152    fn open_checked_out(&self) -> Result<PooledReadConnection> {
1153        match self.inner.open_reader() {
1154            Ok(conn) => {
1155                let mut state = lock_recover(&self.inner.state);
1156                if state.closed {
1157                    drop(conn);
1158                    state.total -= 1;
1159                    self.inner.drained.notify_all();
1160                    return Err(DbError::ReadPoolClosed);
1161                }
1162                Ok(PooledReadConnection {
1163                    conn: Some(conn),
1164                    pool: Arc::downgrade(&self.inner),
1165                    _lifecycle_lease: None,
1166                })
1167            }
1168            Err(error) => {
1169                let mut state = lock_recover(&self.inner.state);
1170                state.total -= 1;
1171                self.inner.available.notify_one();
1172                self.inner.drained.notify_all();
1173                Err(error)
1174            }
1175        }
1176    }
1177
1178    /// Stops handing out readers and drops the idle ones now. A reader still checked out closes
1179    /// when it is returned. Calling it again does nothing.
1180    pub fn close(&self) {
1181        let idle = {
1182            let mut state = lock_recover(&self.inner.state);
1183            if state.closed {
1184                return;
1185            }
1186            state.closed = true;
1187            self.inner.available.notify_all();
1188            std::mem::take(&mut state.idle)
1189        };
1190        let closed = idle.len();
1191        // Files close before the count drops, so `close_and_wait` sees them released.
1192        drop(idle);
1193        let mut state = lock_recover(&self.inner.state);
1194        state.total -= closed;
1195        self.inner.drained.notify_all();
1196    }
1197
1198    /// [`Self::close`], then waits up to `timeout` for every checked-out reader to be returned.
1199    /// `Ok(())` means every connection is closed and its database file released. Otherwise it
1200    /// returns [`DbError::ReadPoolCloseTimeout`] with the number still out. Do not call it while
1201    /// holding a reader from this pool: that reader cannot come back, so the call times out. A
1202    /// timeout too large for the clock waits in bounded slices until the readers return.
1203    pub fn close_and_wait(&self, timeout: Duration) -> Result<()> {
1204        self.close();
1205        let deadline = Instant::now().checked_add(timeout);
1206        let mut state = lock_recover(&self.inner.state);
1207        loop {
1208            if state.total == 0 {
1209                return Ok(());
1210            }
1211            let remaining = match deadline {
1212                Some(at) => at.saturating_duration_since(Instant::now()),
1213                None => Duration::from_secs(60),
1214            };
1215            if remaining.is_zero() {
1216                return Err(DbError::ReadPoolCloseTimeout {
1217                    outstanding: state.total,
1218                    timeout,
1219                });
1220            }
1221            let (next, _) = match self.inner.drained.wait_timeout(state, remaining) {
1222                Ok(pair) => pair,
1223                Err(poisoned) => poisoned.into_inner(),
1224            };
1225            state = next;
1226        }
1227    }
1228
1229    pub fn capacity(&self) -> usize {
1230        self.inner.cap
1231    }
1232
1233    pub fn idle_len(&self) -> usize {
1234        lock_recover(&self.inner.state).idle.len()
1235    }
1236
1237    pub fn total_len(&self) -> usize {
1238        lock_recover(&self.inner.state).total
1239    }
1240}
1241
1242fn lock_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
1243    mutex
1244        .lock()
1245        .unwrap_or_else(|poisoned| poisoned.into_inner())
1246}
1247
1248#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1249pub struct CheckpointStatus {
1250    pub busy: i64,
1251    pub log_frames: i64,
1252    pub checkpointed_frames: i64,
1253}
1254
1255pub fn checkpoint_truncate(conn: &Connection) -> Result<CheckpointStatus> {
1256    checkpoint(conn, "PRAGMA wal_checkpoint(TRUNCATE)")
1257}
1258
1259/// `PRAGMA wal_checkpoint(RESTART)`: copies the log into the main file like PASSIVE, waits (under
1260/// the busy timeout) for readers still using the log, then the next writer starts it again from
1261/// the beginning. The log file keeps its size. Use it when the database is quiet and the log is
1262/// large; [`WalCheckpointJob`] decides when.
1263pub fn checkpoint_restart(conn: &Connection) -> Result<CheckpointStatus> {
1264    checkpoint(conn, "PRAGMA wal_checkpoint(RESTART)")
1265}
1266
1267pub fn checkpoint_if_wal_larger_than(
1268    conn: &Connection,
1269    db: &Path,
1270    bytes: u64,
1271) -> Result<Option<CheckpointStatus>> {
1272    let wal = wal_path(db)?;
1273    let size = match std::fs::metadata(wal) {
1274        Ok(metadata) => metadata.len(),
1275        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
1276        Err(_) => return Ok(None),
1277    };
1278    if size <= bytes {
1279        return Ok(None);
1280    }
1281    checkpoint(conn, "PRAGMA wal_checkpoint(PASSIVE)").map(Some)
1282}
1283
1284fn checkpoint(conn: &Connection, pragma: &str) -> Result<CheckpointStatus> {
1285    conn.query_row(pragma, [], |row| {
1286        Ok(CheckpointStatus {
1287            busy: row.get(0)?,
1288            log_frames: row.get(1)?,
1289            checkpointed_frames: row.get(2)?,
1290        })
1291    })
1292    .map_err(Into::into)
1293}
1294
1295fn wal_path(db: &Path) -> Result<PathBuf> {
1296    let file_name = db.file_name().ok_or(DbError::InvalidDatabasePath)?;
1297    let mut wal_name = OsString::from(file_name);
1298    wal_name.push("-wal");
1299    Ok(db.with_file_name(wal_name))
1300}
1301
1302pub fn optimize_on_open(conn: &Connection) -> Result<()> {
1303    conn.execute_batch("PRAGMA optimize=0x10002;")?;
1304    Ok(())
1305}
1306
1307pub fn optimize_boundary(conn: &Connection) -> Result<()> {
1308    conn.execute_batch("PRAGMA optimize;")?;
1309    Ok(())
1310}