1use rusqlite::{
9 params, Connection, ErrorCode, OptionalExtension, Transaction, TransactionBehavior,
10};
11use std::ffi::OsString;
12use std::fmt;
13use std::fs::{File, OpenOptions, TryLockError};
14use std::path::PathBuf;
15use std::sync::Mutex;
16use std::time::Duration;
17
18mod diagnostic;
19pub use diagnostic::{failure_diagnostic, FailureDiagnostic};
20
21const SCHEMA_MARKER_TABLE: &str = "_harn_sqlite_schema_versions";
22const CREATE_SCHEMA_MARKER_TABLE: &str =
23 "CREATE TABLE IF NOT EXISTS main._harn_sqlite_schema_versions (
24 name TEXT PRIMARY KEY,
25 version INTEGER NOT NULL CHECK(version > 0)
26);";
27static TRANSIENT_INITIALIZATION_LOCK: Mutex<()> = Mutex::new(());
28
29#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub struct SchemaVersion {
32 name: &'static str,
33 version: i64,
34}
35
36#[derive(Clone, Copy, Debug, PartialEq, Eq)]
38pub enum SqliteContention {
39 Busy,
41 Locked,
43}
44
45#[must_use]
47pub fn sqlite_contention(error: &rusqlite::Error) -> Option<SqliteContention> {
48 match error {
49 rusqlite::Error::SqliteFailure(failure, _) => match failure.code {
50 ErrorCode::DatabaseBusy => Some(SqliteContention::Busy),
51 ErrorCode::DatabaseLocked => Some(SqliteContention::Locked),
52 _ => None,
53 },
54 _ => None,
55 }
56}
57
58impl SchemaVersion {
59 #[must_use]
63 pub const fn new(name: &'static str, version: i64) -> Self {
64 assert!(!name.is_empty(), "SQLite schema name must not be empty");
65 assert!(version > 0, "SQLite schema version must be positive");
66 Self { name, version }
67 }
68}
69
70#[derive(Debug)]
72#[non_exhaustive]
73pub enum InitializationError<E> {
74 BusyTimeoutTooLarge { milliseconds: u128 },
76 BusyTimeout(rusqlite::Error),
78 JournalModeQuery(rusqlite::Error),
80 DatabasePathUnavailable,
82 FileBackedTransient { path: PathBuf },
84 DatabasePath {
86 path: PathBuf,
87 source: std::io::Error,
88 },
89 InitializationLockOpen {
91 path: PathBuf,
92 source: std::io::Error,
93 },
94 InitializationLockAcquire {
96 path: PathBuf,
97 source: std::io::Error,
98 },
99 InitializationLock(harn_flock::LockError),
102 WalPragma(rusqlite::Error),
104 WalNotEnabled { mode: String },
106 WalBusyNotWal { mode: String },
108 WalBusyQuery {
110 wal_error: Box<rusqlite::Error>,
111 query_error: Box<rusqlite::Error>,
112 },
113 Synchronous(rusqlite::Error),
115 SchemaReadiness(rusqlite::Error),
117 SchemaNotInitialized { name: &'static str, version: i64 },
120 NewerSchemaVersion {
122 name: &'static str,
123 stored: i64,
124 supported: i64,
125 },
126 Transaction(rusqlite::Error),
128 Initialize(E),
130 SchemaMarker(rusqlite::Error),
132 Commit(rusqlite::Error),
134}
135
136impl InitializationError<rusqlite::Error> {
137 #[must_use]
139 pub fn is_busy_or_locked(&self) -> bool {
140 if let Self::Initialize(error) = self {
141 is_sqlite_busy_or_locked(error)
142 } else {
143 initialization_stage_is_busy_or_locked(self)
144 }
145 }
146}
147
148impl<E: fmt::Display> fmt::Display for InitializationError<E> {
149 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
150 match self {
151 Self::BusyTimeoutTooLarge { milliseconds } => write!(
152 f,
153 "busy_timeout {milliseconds}ms exceeds SQLite's maximum"
154 ),
155 Self::BusyTimeout(error) => write!(f, "busy_timeout failed: {error}"),
156 Self::JournalModeQuery(error) => write!(f, "journal_mode query failed: {error}"),
157 Self::DatabasePathUnavailable => {
158 write!(f, "SQLite connection has no file-backed main database path")
159 }
160 Self::FileBackedTransient { path } => write!(
161 f,
162 "transient SQLite initialization requires a private non-file database, got {}",
163 path.display()
164 ),
165 Self::DatabasePath { path, source } => write!(
166 f,
167 "could not resolve SQLite database path {}: {source}",
168 path.display()
169 ),
170 Self::InitializationLockOpen { path, source } => write!(
171 f,
172 "could not open SQLite initialization lock {}: {source}",
173 path.display()
174 ),
175 Self::InitializationLockAcquire { path, source } => write!(
176 f,
177 "could not acquire SQLite initialization lock {}: {source}",
178 path.display()
179 ),
180 Self::InitializationLock(error) => {
181 write!(f, "SQLite initialization lock unavailable: {error}")
182 }
183 Self::WalPragma(error) => write!(f, "WAL journal_mode pragma failed: {error}"),
184 Self::WalNotEnabled { mode } => {
185 write!(f, "WAL journal_mode request returned {mode}")
186 }
187 Self::WalBusyNotWal { mode } => {
188 write!(f, "WAL journal_mode request left journal_mode at {mode}")
189 }
190 Self::WalBusyQuery {
191 wal_error,
192 query_error,
193 } => write!(
194 f,
195 "WAL journal_mode pragma failed: {wal_error}; journal_mode query also failed: {query_error}"
196 ),
197 Self::Synchronous(error) => write!(f, "synchronous pragma failed: {error}"),
198 Self::SchemaReadiness(error) => write!(f, "schema readiness query failed: {error}"),
199 Self::SchemaNotInitialized { name, version } => {
200 write!(f, "SQLite schema {name} version {version} is not initialized")
201 }
202 Self::NewerSchemaVersion {
203 name,
204 stored,
205 supported,
206 } => write!(
207 f,
208 "SQLite schema {name} version {stored} is newer than supported version {supported}"
209 ),
210 Self::Transaction(error) => write!(f, "schema transaction failed: {error}"),
211 Self::Initialize(error) => write!(f, "schema initialization failed: {error}"),
212 Self::SchemaMarker(error) => write!(f, "schema marker update failed: {error}"),
213 Self::Commit(error) => write!(f, "schema transaction commit failed: {error}"),
214 }
215 }
216}
217
218impl<E: std::error::Error + 'static> std::error::Error for InitializationError<E> {
219 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
220 match self {
221 Self::BusyTimeout(error)
222 | Self::JournalModeQuery(error)
223 | Self::WalPragma(error)
224 | Self::Synchronous(error)
225 | Self::SchemaReadiness(error)
226 | Self::Transaction(error)
227 | Self::SchemaMarker(error)
228 | Self::Commit(error) => Some(error),
229 Self::DatabasePath { source, .. }
230 | Self::InitializationLockOpen { source, .. }
231 | Self::InitializationLockAcquire { source, .. } => Some(source),
232 Self::InitializationLock(error) => Some(error),
233 Self::WalBusyQuery { wal_error, .. } => Some(wal_error),
234 Self::Initialize(error) => Some(error),
235 Self::BusyTimeoutTooLarge { .. }
236 | Self::DatabasePathUnavailable
237 | Self::FileBackedTransient { .. }
238 | Self::SchemaNotInitialized { .. }
239 | Self::WalNotEnabled { .. }
240 | Self::WalBusyNotWal { .. }
241 | Self::NewerSchemaVersion { .. } => None,
242 }
243 }
244}
245
246impl<E> InitializationError<E> {
247 #[must_use]
249 pub fn stage(&self) -> &'static str {
250 match self {
251 Self::BusyTimeoutTooLarge { .. } | Self::BusyTimeout(_) => "busy_timeout",
252 Self::JournalModeQuery(_) => "journal_mode_query",
253 Self::DatabasePathUnavailable
254 | Self::DatabasePath { .. }
255 | Self::FileBackedTransient { .. } => "database_path",
256 Self::InitializationLockOpen { .. } => "initialization_lock_open",
257 Self::InitializationLockAcquire { .. } | Self::InitializationLock(_) => {
258 "initialization_lock_acquire"
259 }
260 Self::WalPragma(_)
261 | Self::WalNotEnabled { .. }
262 | Self::WalBusyNotWal { .. }
263 | Self::WalBusyQuery { .. } => "wal_promotion",
264 Self::Synchronous(_) => "synchronous",
265 Self::SchemaReadiness(_)
266 | Self::SchemaNotInitialized { .. }
267 | Self::NewerSchemaVersion { .. } => "schema_readiness",
268 Self::Transaction(_) => "schema_transaction",
269 Self::Initialize(_) => "schema_initialize",
270 Self::SchemaMarker(_) => "schema_marker",
271 Self::Commit(_) => "schema_commit",
272 }
273 }
274}
275
276pub fn initialize_file<E, F>(
287 connection: &Connection,
288 busy_timeout: Duration,
289 schema: SchemaVersion,
290 initialize: F,
291) -> Result<(), InitializationError<E>>
292where
293 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
294{
295 configure_busy_timeout(connection, busy_timeout)?;
296 if fast_path_is_ready(connection, schema)? {
297 return configure_connection(connection);
298 }
299
300 let _initialization_lock = acquire_initialization_lock(connection, lock_timeout())?;
301 ensure_wal_journal_mode(connection)?;
302 configure_connection(connection)?;
303 initialize_schema(connection, schema, initialize)
304}
305
306fn fast_path_is_ready<E>(
307 connection: &Connection,
308 schema: SchemaVersion,
309) -> Result<bool, InitializationError<E>> {
310 match is_wal_journal_mode(connection) {
311 Ok(true) => {}
312 Ok(false) => return Ok(false),
313 Err(error) if initialization_stage_is_busy_or_locked(&error) => return Ok(false),
314 Err(error) => return Err(error),
315 }
316 match schema_is_ready(connection, schema) {
317 Ok(ready) => Ok(ready),
318 Err(error) if initialization_stage_is_busy_or_locked(&error) => Ok(false),
319 Err(error) => Err(error),
320 }
321}
322
323fn initialization_stage_is_busy_or_locked<E>(error: &InitializationError<E>) -> bool {
324 match error {
325 InitializationError::BusyTimeout(error)
326 | InitializationError::JournalModeQuery(error)
327 | InitializationError::WalPragma(error)
328 | InitializationError::Synchronous(error)
329 | InitializationError::SchemaReadiness(error)
330 | InitializationError::Transaction(error)
331 | InitializationError::SchemaMarker(error)
332 | InitializationError::Commit(error) => is_sqlite_busy_or_locked(error),
333 InitializationError::WalBusyNotWal { .. } => true,
334 InitializationError::WalBusyQuery {
335 wal_error,
336 query_error,
337 } => is_sqlite_busy_or_locked(wal_error) || is_sqlite_busy_or_locked(query_error),
338 InitializationError::BusyTimeoutTooLarge { .. }
339 | InitializationError::DatabasePath { .. }
340 | InitializationError::DatabasePathUnavailable
341 | InitializationError::FileBackedTransient { .. }
342 | InitializationError::InitializationLockOpen { .. }
343 | InitializationError::InitializationLockAcquire { .. }
344 | InitializationError::InitializationLock(_)
345 | InitializationError::SchemaNotInitialized { .. }
346 | InitializationError::NewerSchemaVersion { .. }
347 | InitializationError::WalNotEnabled { .. }
348 | InitializationError::Initialize(_) => false,
349 }
350}
351
352pub fn require_file_initialized<E>(
361 connection: &Connection,
362 busy_timeout: Duration,
363 schema: SchemaVersion,
364) -> Result<(), InitializationError<E>> {
365 require_file_initialized_impl(connection, busy_timeout, schema, || {})
366}
367
368pub fn require_snapshot_initialized<E>(
377 connection: &Connection,
378 schema: SchemaVersion,
379) -> Result<(), InitializationError<E>> {
380 if schema_is_ready(connection, schema)? {
381 Ok(())
382 } else {
383 Err(InitializationError::SchemaNotInitialized {
384 name: schema.name,
385 version: schema.version,
386 })
387 }
388}
389
390fn require_file_initialized_impl<E>(
391 connection: &Connection,
392 busy_timeout: Duration,
393 schema: SchemaVersion,
394 on_readiness_contention: impl FnOnce(),
395) -> Result<(), InitializationError<E>> {
396 configure_busy_timeout(connection, busy_timeout)?;
397 if fast_path_is_ready(connection, schema)? {
398 return Ok(());
399 }
400
401 let _readiness_lock =
402 acquire_readiness_lock(connection, schema, lock_timeout(), on_readiness_contention)?;
403 if is_wal_journal_mode(connection)? && schema_is_ready(connection, schema)? {
404 return Ok(());
405 }
406 Err(InitializationError::SchemaNotInitialized {
407 name: schema.name,
408 version: schema.version,
409 })
410}
411
412pub fn initialize_transient<E, F>(
421 connection: &Connection,
422 busy_timeout: Duration,
423 schema: SchemaVersion,
424 initialize: F,
425) -> Result<(), InitializationError<E>>
426where
427 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
428{
429 configure_busy_timeout(connection, busy_timeout)?;
430 if let Some(path) = main_database_path(connection) {
431 return Err(InitializationError::FileBackedTransient { path });
432 }
433 let _initialization_lock = TRANSIENT_INITIALIZATION_LOCK
434 .lock()
435 .unwrap_or_else(std::sync::PoisonError::into_inner);
436 configure_connection(connection)?;
437 if schema_is_ready(connection, schema)? {
438 return Ok(());
439 }
440 initialize_schema(connection, schema, initialize)
441}
442
443fn initialize_schema<E, F>(
444 connection: &Connection,
445 schema: SchemaVersion,
446 initialize: F,
447) -> Result<(), InitializationError<E>>
448where
449 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
450{
451 let transaction = Transaction::new_unchecked(connection, TransactionBehavior::Immediate)
452 .map_err(InitializationError::Transaction)?;
453 transaction
454 .execute_batch(CREATE_SCHEMA_MARKER_TABLE)
455 .map_err(InitializationError::SchemaMarker)?;
456 if schema_marker_is_ready(&transaction, schema)? {
457 return transaction.commit().map_err(InitializationError::Commit);
458 }
459 initialize(&transaction).map_err(InitializationError::Initialize)?;
460 transaction
461 .execute(
462 "INSERT INTO main._harn_sqlite_schema_versions(name, version) VALUES (?1, ?2)
463 ON CONFLICT(name) DO UPDATE SET version = excluded.version",
464 params![schema.name, schema.version],
465 )
466 .map_err(InitializationError::SchemaMarker)?;
467 transaction.commit().map_err(InitializationError::Commit)
468}
469
470fn configure_busy_timeout<E>(
471 connection: &Connection,
472 busy_timeout: Duration,
473) -> Result<(), InitializationError<E>> {
474 let milliseconds = busy_timeout.as_millis();
475 if milliseconds > i32::MAX as u128 {
476 return Err(InitializationError::BusyTimeoutTooLarge { milliseconds });
477 }
478 connection
479 .busy_timeout(busy_timeout)
480 .map_err(InitializationError::BusyTimeout)
481}
482
483fn configure_connection<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
484 connection
485 .pragma_update(None, "synchronous", "NORMAL")
486 .map_err(InitializationError::Synchronous)
487}
488
489pub const LOCK_TIMEOUT_SECONDS_ENV: &str = "HARN_SQLITE_LOCK_TIMEOUT_SECONDS";
491
492const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
499
500fn lock_timeout() -> Duration {
507 std::env::var(LOCK_TIMEOUT_SECONDS_ENV)
508 .ok()
509 .and_then(|value| value.trim().parse::<u64>().ok())
510 .filter(|seconds| *seconds > 0)
511 .map_or(DEFAULT_LOCK_TIMEOUT, Duration::from_secs)
512}
513
514fn acquire_initialization_lock<E>(
515 connection: &Connection,
516 timeout: Duration,
517) -> Result<SqliteInitializationLock, InitializationError<E>> {
518 let path = initialization_lock_path(connection)?;
519 let file = OpenOptions::new()
520 .create(true)
521 .truncate(false)
522 .read(true)
523 .write(true)
524 .open(&path)
525 .map_err(|source| InitializationError::InitializationLockOpen {
526 path: path.clone(),
527 source,
528 })?;
529 harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Exclusive, timeout)
530 .map_err(InitializationError::InitializationLock)?;
531 Ok(SqliteInitializationLock { file })
532}
533
534fn acquire_readiness_lock<E>(
535 connection: &Connection,
536 schema: SchemaVersion,
537 timeout: Duration,
538 on_contention: impl FnOnce(),
539) -> Result<SqliteInitializationLock, InitializationError<E>> {
540 let path = initialization_lock_path(connection)?;
541 let file = match OpenOptions::new().read(true).open(&path) {
542 Ok(file) => file,
543 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
544 return Err(InitializationError::SchemaNotInitialized {
545 name: schema.name,
546 version: schema.version,
547 });
548 }
549 Err(source) => {
550 return Err(InitializationError::InitializationLockOpen { path, source });
551 }
552 };
553 match file.try_lock_shared() {
554 Ok(()) => {}
555 Err(TryLockError::WouldBlock) => {
556 on_contention();
557 harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Shared, timeout)
558 .map_err(InitializationError::InitializationLock)?;
559 }
560 Err(TryLockError::Error(source)) => {
561 return Err(InitializationError::InitializationLockAcquire { path, source });
562 }
563 }
564 Ok(SqliteInitializationLock { file })
565}
566
567fn initialization_lock_path<E>(connection: &Connection) -> Result<PathBuf, InitializationError<E>> {
568 let database_path =
569 main_database_path(connection).ok_or(InitializationError::DatabasePathUnavailable)?;
570 let canonical = std::fs::canonicalize(&database_path).map_err(|source| {
571 InitializationError::DatabasePath {
572 path: database_path.clone(),
573 source,
574 }
575 })?;
576 let mut path = OsString::from(canonical.as_os_str());
577 path.push(".harn-init.lock");
578 Ok(PathBuf::from(path))
579}
580
581#[cfg(unix)]
582fn main_database_path(connection: &Connection) -> Option<PathBuf> {
583 use std::ffi::{CStr, OsStr};
584 use std::os::unix::ffi::OsStrExt;
585
586 let filename = unsafe {
589 let pointer =
590 rusqlite::ffi::sqlite3_db_filename(connection.handle(), rusqlite::MAIN_DB.as_ptr());
591 (!pointer.is_null()).then(|| CStr::from_ptr(pointer).to_bytes())
592 }?;
593 (!filename.is_empty()).then(|| PathBuf::from(OsStr::from_bytes(filename)))
594}
595
596#[cfg(not(unix))]
597fn main_database_path(connection: &Connection) -> Option<PathBuf> {
598 connection
599 .path()
600 .filter(|path| !path.is_empty())
601 .map(PathBuf::from)
602}
603
604struct SqliteInitializationLock {
605 file: File,
606}
607
608impl Drop for SqliteInitializationLock {
609 fn drop(&mut self) {
610 let _ = self.file.unlock();
613 }
614}
615
616fn schema_is_ready<E>(
617 connection: &Connection,
618 schema: SchemaVersion,
619) -> Result<bool, InitializationError<E>> {
620 let marker_exists = connection
621 .query_row(
622 "SELECT EXISTS(
623 SELECT 1 FROM main.sqlite_schema WHERE type = 'table' AND name = ?1
624 )",
625 params![SCHEMA_MARKER_TABLE],
626 |row| row.get::<_, bool>(0),
627 )
628 .map_err(InitializationError::SchemaReadiness)?;
629 if !marker_exists {
630 return Ok(false);
631 }
632 schema_marker_is_ready(connection, schema)
633}
634
635fn schema_marker_is_ready<E>(
636 connection: &Connection,
637 schema: SchemaVersion,
638) -> Result<bool, InitializationError<E>> {
639 let stored = connection
640 .query_row(
641 "SELECT version FROM main._harn_sqlite_schema_versions WHERE name = ?1",
642 params![schema.name],
643 |row| row.get::<_, i64>(0),
644 )
645 .optional()
646 .map_err(InitializationError::SchemaReadiness)?;
647 match stored {
648 Some(version) if version > schema.version => Err(InitializationError::NewerSchemaVersion {
649 name: schema.name,
650 stored: version,
651 supported: schema.version,
652 }),
653 Some(version) => Ok(version == schema.version),
654 None => Ok(false),
655 }
656}
657
658fn is_wal_journal_mode<E>(connection: &Connection) -> Result<bool, InitializationError<E>> {
659 current_journal_mode(connection)
660 .map(|mode| mode.eq_ignore_ascii_case("wal"))
661 .map_err(InitializationError::JournalModeQuery)
662}
663
664fn ensure_wal_journal_mode<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
665 if is_wal_journal_mode(connection)? {
666 return Ok(());
667 }
668 match connection.query_row("PRAGMA journal_mode = WAL", [], |row| {
669 row.get::<_, String>(0)
670 }) {
671 Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
672 Ok(mode) => Err(InitializationError::WalNotEnabled { mode }),
673 Err(error) if is_sqlite_busy_or_locked(&error) => match current_journal_mode(connection) {
674 Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
675 Ok(mode) => Err(InitializationError::WalBusyNotWal { mode }),
676 Err(query_error) => Err(InitializationError::WalBusyQuery {
677 wal_error: Box::new(error),
678 query_error: Box::new(query_error),
679 }),
680 },
681 Err(error) => Err(InitializationError::WalPragma(error)),
682 }
683}
684
685fn current_journal_mode(connection: &Connection) -> Result<String, rusqlite::Error> {
686 connection.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))
687}
688
689fn is_sqlite_busy_or_locked(error: &rusqlite::Error) -> bool {
690 sqlite_contention(error).is_some()
691}
692
693#[cfg(test)]
694#[path = "tests.rs"]
695mod tests;