1#![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};
14pub use zeroize::Zeroizing;
17
18mod backup;
19mod checkpoint_job;
20#[allow(unsafe_code)]
21mod host;
22pub use host::per_host_path;
23pub 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#[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 #[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 #[error("read pool is closed")]
96 ReadPoolClosed,
97 #[error("read pool close timed out after {timeout:?} with {outstanding} reader(s) still out")]
100 ReadPoolCloseTimeout {
101 outstanding: usize,
102 timeout: Duration,
103 },
104 #[error("reader connection is not read-only: {0}")]
107 ReadOnlyNotEffective(String),
108 #[error("database path has no file name")]
109 InvalidDatabasePath,
110 #[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 #[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 #[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_failures: u32,
159 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#[derive(Clone, Copy, Debug, Eq, PartialEq)]
182#[non_exhaustive]
183pub enum JournalSelect {
184 WalWithFallback,
187 WalStrict,
189 Delete,
191 Truncate,
193}
194
195impl JournalSelect {
196 fn is_rollback(self) -> bool {
197 matches!(self, JournalSelect::Delete | JournalSelect::Truncate)
198 }
199}
200
201pub const JOURNAL_ENV: &str = "RIGHTKIT_SQLITE_JOURNAL_MODE";
204
205#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
208pub enum IntegrityCheck {
209 #[default]
211 Off,
212 Quick,
215 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 pub busy_timeout: Duration,
227 pub key: Option<Zeroizing<String>>,
229 pub raw_key: Option<Zeroizing<[u8; 32]>>,
231 pub profile: CacheProfile,
232 pub durability: DurabilityProfile,
233 pub foreign_keys: bool,
234 pub integrity_check_on_open: bool,
236 pub integrity: IntegrityCheck,
239 pub journal: JournalSelect,
240 pub owner_only: bool,
244 pub init: Option<ConnInit>,
245 pub journal_env: Option<String>,
248 pub per_host_fallback: bool,
253 pub secure_delete: Option<bool>,
256 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 pub fn without_optimize_on_open(mut self) -> Self {
296 self.optimize_on_open = false;
297 self
298 }
299
300 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 pub fn open_read_only(&self) -> Result<Connection> {
343 self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_ONLY)
344 }
345
346 pub fn open_existing_writer(&self) -> Result<Connection> {
349 self.open_existing_at(&self.path, OpenFlags::SQLITE_OPEN_READ_WRITE)
350 }
351
352 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 pub fn open_read_pool(&self, cap: usize) -> Result<ReadPool> {
391 ReadPool::open(self.clone(), cap, ReadPoolOptions::new())
392 }
393
394 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 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 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 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 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 let _pin = pin_store_path(path, true, true)?;
482 let conn = open_store(path, flags)?;
483 conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, true)?;
488 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 conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, false)?;
503 Ok((conn, journal))
504 }
505
506 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 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
542pub fn open_read_only(path: impl AsRef<Path>) -> Result<Connection> {
544 DbOptions::new(path.as_ref()).open_read_only()
545}
546
547pub fn open_existing_writer(path: impl AsRef<Path>) -> Result<Connection> {
549 DbOptions::new(path.as_ref()).open_existing_writer()
550}
551
552pub fn open_in_memory() -> Result<Connection> {
554 DbOptions::new(":memory:").open_in_memory()
555}
556
557fn 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#[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#[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 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#[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 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#[derive(Clone, Copy, Debug, Eq, PartialEq)]
680enum Opened {
681 AsRequested,
682 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 enforce_query_only(conn)?;
712 } else if options.optimize_on_open {
713 optimize_on_open(conn)?;
714 }
715 Ok(opened)
716}
717
718fn 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
737fn 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 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 && (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
795fn fallback_mode(options: &DbOptions) -> JournalSelect {
798 if options.per_host_fallback {
799 JournalSelect::Truncate
800 } else {
801 JournalSelect::Delete
802 }
803}
804
805fn 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
840pub fn journal_mode(conn: &Connection) -> Result<String> {
842 Ok(conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))?)
843}
844
845pub fn write_txn(conn: &mut Connection) -> Result<Transaction<'_>> {
848 conn.transaction_with_behavior(TransactionBehavior::Immediate)
849 .map_err(Into::into)
850}
851
852pub 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#[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 pub fn with_read_only(mut self, on: bool) -> Self {
936 self.read_only = on;
937 self
938 }
939
940 pub fn with_reader_cache_kib(mut self, kib: u32) -> Self {
946 self.cache_kib = Some(kib);
947 self
948 }
949}
950
951#[derive(Clone)]
953pub struct ReadPool {
954 inner: Arc<ReadPoolInner>,
955}
956
957type ReaderOpener = Box<dyn Fn() -> Result<Connection> + Send + Sync>;
960
961struct ReadPoolInner {
962 open: ReaderOpener,
963 cap: usize,
964 state: Mutex<ReadPoolState>,
965 available: Condvar,
967 drained: Condvar,
969}
970
971#[derive(Default)]
972struct ReadPoolState {
973 idle: Vec<Connection>,
974 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 _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 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 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 pub fn get(&self) -> Result<PooledReadConnection> {
1083 match self.checkout(None)? {
1084 Some(reader) => Ok(reader),
1085 None => Err(DbError::ReadPoolClosed),
1087 }
1088 }
1089
1090 pub fn get_timeout(&self, timeout: Duration) -> Result<Option<PooledReadConnection>> {
1097 self.checkout(Some(timeout))
1098 }
1099
1100 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 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 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 drop(idle);
1193 let mut state = lock_recover(&self.inner.state);
1194 state.total -= closed;
1195 self.inner.drained.notify_all();
1196 }
1197
1198 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
1259pub 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}