harn_sqlite/
lib.rs

1//! Shared initialization for Harn-owned SQLite databases.
2//!
3//! File-backed databases use a persistent sidecar advisory lock while changing
4//! journal mode and committing a versioned schema marker. Once both WAL and the
5//! exact marker are visible, later opens avoid the lock. Transient databases
6//! share the schema transaction contract without a filesystem lock or WAL.
7
8use 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 mod schema;
20pub use diagnostic::{failure_diagnostic, FailureDiagnostic};
21
22const SCHEMA_MARKER_TABLE: &str = "_harn_sqlite_schema_versions";
23const CREATE_SCHEMA_MARKER_TABLE: &str =
24    "CREATE TABLE IF NOT EXISTS main._harn_sqlite_schema_versions (
25    name TEXT PRIMARY KEY,
26    version INTEGER NOT NULL CHECK(version > 0)
27);";
28static TRANSIENT_INITIALIZATION_LOCK: Mutex<()> = Mutex::new(());
29
30/// Stable identity for one schema stored in a Harn-owned SQLite database.
31#[derive(Clone, Copy, Debug, PartialEq, Eq)]
32pub struct SchemaVersion {
33    name: &'static str,
34    version: i64,
35}
36
37/// Lock-contention reason reported by SQLite.
38#[derive(Clone, Copy, Debug, PartialEq, Eq)]
39pub enum SqliteContention {
40    /// Another connection currently owns the database write lock.
41    Busy,
42    /// A shared-cache table lock prevents the operation from proceeding.
43    Locked,
44}
45
46/// Classify a SQLite error without relying on rendered error text.
47#[must_use]
48pub fn sqlite_contention(error: &rusqlite::Error) -> Option<SqliteContention> {
49    match error {
50        rusqlite::Error::SqliteFailure(failure, _) => match failure.code {
51            ErrorCode::DatabaseBusy => Some(SqliteContention::Busy),
52            ErrorCode::DatabaseLocked => Some(SqliteContention::Locked),
53            _ => None,
54        },
55        _ => None,
56    }
57}
58
59impl SchemaVersion {
60    /// Define a non-empty schema name and positive version.
61    ///
62    /// Invalid constants fail during compile-time evaluation.
63    #[must_use]
64    pub const fn new(name: &'static str, version: i64) -> Self {
65        assert!(!name.is_empty(), "SQLite schema name must not be empty");
66        assert!(version > 0, "SQLite schema version must be positive");
67        Self { name, version }
68    }
69}
70
71/// Failure to configure or initialize a Harn-owned SQLite database.
72#[derive(Debug)]
73#[non_exhaustive]
74pub enum InitializationError<E> {
75    /// The configured busy timeout cannot be represented by SQLite.
76    BusyTimeoutTooLarge { milliseconds: u128 },
77    /// The connection rejected its configured busy timeout.
78    BusyTimeout(rusqlite::Error),
79    /// The current journal mode could not be observed.
80    JournalModeQuery(rusqlite::Error),
81    /// The connection does not own a file-backed main database.
82    DatabasePathUnavailable,
83    /// A file-backed connection was passed to the transient initializer.
84    FileBackedTransient { path: PathBuf },
85    /// The opened database path could not be resolved to one lock identity.
86    DatabasePath {
87        path: PathBuf,
88        source: std::io::Error,
89    },
90    /// The persistent sidecar lock could not be opened.
91    InitializationLockOpen {
92        path: PathBuf,
93        source: std::io::Error,
94    },
95    /// Exclusive ownership of the sidecar lock could not be acquired.
96    InitializationLockAcquire {
97        path: PathBuf,
98        source: std::io::Error,
99    },
100    /// The sidecar lock was still held when the wait budget expired, or the
101    /// operating system refused it. Carries the path and the elapsed wait.
102    InitializationLock(harn_flock::LockError),
103    /// SQLite rejected WAL journal mode for a non-contention reason.
104    WalPragma(rusqlite::Error),
105    /// SQLite accepted the journal-mode request but returned another mode.
106    WalNotEnabled { mode: String },
107    /// WAL promotion was busy and the database remained in another mode.
108    WalBusyNotWal { mode: String },
109    /// WAL promotion and the diagnostic journal-mode query both failed.
110    WalBusyQuery {
111        wal_error: Box<rusqlite::Error>,
112        query_error: Box<rusqlite::Error>,
113    },
114    /// The connection rejected the runtime synchronous setting.
115    Synchronous(rusqlite::Error),
116    /// The schema marker could not be inspected.
117    SchemaReadiness(rusqlite::Error),
118    /// No initializer committed the exact schema version before the readiness
119    /// lease became available.
120    SchemaNotInitialized { name: &'static str, version: i64 },
121    /// The database was initialized by a newer incompatible schema owner.
122    NewerSchemaVersion {
123        name: &'static str,
124        stored: i64,
125        supported: i64,
126    },
127    /// The atomic schema transaction could not begin.
128    Transaction(rusqlite::Error),
129    /// The owning schema callback failed.
130    Initialize(E),
131    /// The schema marker table or row could not be written.
132    SchemaMarker(rusqlite::Error),
133    /// The schema and marker transaction could not commit.
134    Commit(rusqlite::Error),
135}
136
137impl InitializationError<rusqlite::Error> {
138    /// Whether the SQLite portion of this failure reports lock contention.
139    #[must_use]
140    pub fn is_busy_or_locked(&self) -> bool {
141        if let Self::Initialize(error) = self {
142            is_sqlite_busy_or_locked(error)
143        } else {
144            initialization_stage_is_busy_or_locked(self)
145        }
146    }
147}
148
149impl<E: fmt::Display> fmt::Display for InitializationError<E> {
150    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
151        match self {
152            Self::BusyTimeoutTooLarge { milliseconds } => write!(
153                f,
154                "busy_timeout {milliseconds}ms exceeds SQLite's maximum"
155            ),
156            Self::BusyTimeout(error) => write!(f, "busy_timeout failed: {error}"),
157            Self::JournalModeQuery(error) => write!(f, "journal_mode query failed: {error}"),
158            Self::DatabasePathUnavailable => {
159                write!(f, "SQLite connection has no file-backed main database path")
160            }
161            Self::FileBackedTransient { path } => write!(
162                f,
163                "transient SQLite initialization requires a private non-file database, got {}",
164                path.display()
165            ),
166            Self::DatabasePath { path, source } => write!(
167                f,
168                "could not resolve SQLite database path {}: {source}",
169                path.display()
170            ),
171            Self::InitializationLockOpen { path, source } => write!(
172                f,
173                "could not open SQLite initialization lock {}: {source}",
174                path.display()
175            ),
176            Self::InitializationLockAcquire { path, source } => write!(
177                f,
178                "could not acquire SQLite initialization lock {}: {source}",
179                path.display()
180            ),
181            Self::InitializationLock(error) => {
182                write!(f, "SQLite initialization lock unavailable: {error}")
183            }
184            Self::WalPragma(error) => write!(f, "WAL journal_mode pragma failed: {error}"),
185            Self::WalNotEnabled { mode } => {
186                write!(f, "WAL journal_mode request returned {mode}")
187            }
188            Self::WalBusyNotWal { mode } => {
189                write!(f, "WAL journal_mode request left journal_mode at {mode}")
190            }
191            Self::WalBusyQuery {
192                wal_error,
193                query_error,
194            } => write!(
195                f,
196                "WAL journal_mode pragma failed: {wal_error}; journal_mode query also failed: {query_error}"
197            ),
198            Self::Synchronous(error) => write!(f, "synchronous pragma failed: {error}"),
199            Self::SchemaReadiness(error) => write!(f, "schema readiness query failed: {error}"),
200            Self::SchemaNotInitialized { name, version } => {
201                write!(f, "SQLite schema {name} version {version} is not initialized")
202            }
203            Self::NewerSchemaVersion {
204                name,
205                stored,
206                supported,
207            } => write!(
208                f,
209                "SQLite schema {name} version {stored} is newer than supported version {supported}"
210            ),
211            Self::Transaction(error) => write!(f, "schema transaction failed: {error}"),
212            Self::Initialize(error) => write!(f, "schema initialization failed: {error}"),
213            Self::SchemaMarker(error) => write!(f, "schema marker update failed: {error}"),
214            Self::Commit(error) => write!(f, "schema transaction commit failed: {error}"),
215        }
216    }
217}
218
219impl<E: std::error::Error + 'static> std::error::Error for InitializationError<E> {
220    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
221        match self {
222            Self::BusyTimeout(error)
223            | Self::JournalModeQuery(error)
224            | Self::WalPragma(error)
225            | Self::Synchronous(error)
226            | Self::SchemaReadiness(error)
227            | Self::Transaction(error)
228            | Self::SchemaMarker(error)
229            | Self::Commit(error) => Some(error),
230            Self::DatabasePath { source, .. }
231            | Self::InitializationLockOpen { source, .. }
232            | Self::InitializationLockAcquire { source, .. } => Some(source),
233            Self::InitializationLock(error) => Some(error),
234            Self::WalBusyQuery { wal_error, .. } => Some(wal_error),
235            Self::Initialize(error) => Some(error),
236            Self::BusyTimeoutTooLarge { .. }
237            | Self::DatabasePathUnavailable
238            | Self::FileBackedTransient { .. }
239            | Self::SchemaNotInitialized { .. }
240            | Self::WalNotEnabled { .. }
241            | Self::WalBusyNotWal { .. }
242            | Self::NewerSchemaVersion { .. } => None,
243        }
244    }
245}
246
247impl<E> InitializationError<E> {
248    /// Stable failure stage, without database paths, SQL, or stored values.
249    #[must_use]
250    pub fn stage(&self) -> &'static str {
251        match self {
252            Self::BusyTimeoutTooLarge { .. } | Self::BusyTimeout(_) => "busy_timeout",
253            Self::JournalModeQuery(_) => "journal_mode_query",
254            Self::DatabasePathUnavailable
255            | Self::DatabasePath { .. }
256            | Self::FileBackedTransient { .. } => "database_path",
257            Self::InitializationLockOpen { .. } => "initialization_lock_open",
258            Self::InitializationLockAcquire { .. } | Self::InitializationLock(_) => {
259                "initialization_lock_acquire"
260            }
261            Self::WalPragma(_)
262            | Self::WalNotEnabled { .. }
263            | Self::WalBusyNotWal { .. }
264            | Self::WalBusyQuery { .. } => "wal_promotion",
265            Self::Synchronous(_) => "synchronous",
266            Self::SchemaReadiness(_)
267            | Self::SchemaNotInitialized { .. }
268            | Self::NewerSchemaVersion { .. } => "schema_readiness",
269            Self::Transaction(_) => "schema_transaction",
270            Self::Initialize(_) => "schema_initialize",
271            Self::SchemaMarker(_) => "schema_marker",
272            Self::Commit(_) => "schema_commit",
273        }
274    }
275}
276
277/// Initialize a file-backed Harn database exactly once per schema version.
278///
279/// The lock identity comes from the connection's VFS-resolved main-database
280/// filename, so callers cannot accidentally serialize a different path.
281/// Initialization serializes WAL promotion and the schema transaction across
282/// processes. The callback and version marker commit atomically. Ready opens
283/// observe WAL plus the exact marker and do not acquire the sidecar lock. The
284/// callback is invoked at most once by this function and should keep all
285/// persistent effects inside the supplied transaction so an error rolls them
286/// back with the marker.
287pub fn initialize_file<E, F>(
288    connection: &Connection,
289    busy_timeout: Duration,
290    schema: SchemaVersion,
291    initialize: F,
292) -> Result<(), InitializationError<E>>
293where
294    F: FnOnce(&Transaction<'_>) -> Result<(), E>,
295{
296    configure_busy_timeout(connection, busy_timeout)?;
297    if fast_path_is_ready(connection, schema)? {
298        return configure_connection(connection);
299    }
300
301    let _initialization_lock = acquire_initialization_lock(connection, lock_timeout())?;
302    ensure_wal_journal_mode(connection)?;
303    configure_connection(connection)?;
304    initialize_schema(connection, schema, initialize)
305}
306
307fn fast_path_is_ready<E>(
308    connection: &Connection,
309    schema: SchemaVersion,
310) -> Result<bool, InitializationError<E>> {
311    match is_wal_journal_mode(connection) {
312        Ok(true) => {}
313        Ok(false) => return Ok(false),
314        Err(error) if initialization_stage_is_busy_or_locked(&error) => return Ok(false),
315        Err(error) => return Err(error),
316    }
317    match schema_is_ready(connection, schema) {
318        Ok(ready) => Ok(ready),
319        Err(error) if initialization_stage_is_busy_or_locked(&error) => Ok(false),
320        Err(error) => Err(error),
321    }
322}
323
324fn initialization_stage_is_busy_or_locked<E>(error: &InitializationError<E>) -> bool {
325    match error {
326        InitializationError::BusyTimeout(error)
327        | InitializationError::JournalModeQuery(error)
328        | InitializationError::WalPragma(error)
329        | InitializationError::Synchronous(error)
330        | InitializationError::SchemaReadiness(error)
331        | InitializationError::Transaction(error)
332        | InitializationError::SchemaMarker(error)
333        | InitializationError::Commit(error) => is_sqlite_busy_or_locked(error),
334        InitializationError::WalBusyNotWal { .. } => true,
335        InitializationError::WalBusyQuery {
336            wal_error,
337            query_error,
338        } => is_sqlite_busy_or_locked(wal_error) || is_sqlite_busy_or_locked(query_error),
339        InitializationError::BusyTimeoutTooLarge { .. }
340        | InitializationError::DatabasePath { .. }
341        | InitializationError::DatabasePathUnavailable
342        | InitializationError::FileBackedTransient { .. }
343        | InitializationError::InitializationLockOpen { .. }
344        | InitializationError::InitializationLockAcquire { .. }
345        | InitializationError::InitializationLock(_)
346        | InitializationError::SchemaNotInitialized { .. }
347        | InitializationError::NewerSchemaVersion { .. }
348        | InitializationError::WalNotEnabled { .. }
349        | InitializationError::Initialize(_) => false,
350    }
351}
352
353/// Require an exact schema version on a file-backed connection without
354/// initializing it.
355///
356/// Ready databases avoid the sidecar lock. When initialization is in flight,
357/// this waits for its persistent lease and validates the marker after the
358/// writer releases it. If no initializer owns the lease, an absent marker is
359/// reported here instead of allowing a read-only consumer to fail later on a
360/// missing application table.
361pub fn require_file_initialized<E>(
362    connection: &Connection,
363    busy_timeout: Duration,
364    schema: SchemaVersion,
365) -> Result<(), InitializationError<E>> {
366    require_file_initialized_impl(connection, busy_timeout, schema, || {})
367}
368
369/// Require the exact schema marker on an inert SQLite snapshot.
370///
371/// Unlike [`require_file_initialized`], this does not inspect journal mode,
372/// configure the connection, or acquire the initialization lock. Callers must
373/// first isolate the source from SQLite's durable bookkeeping (for example,
374/// by opening an `immutable=1` URI or a disposable copy). Such snapshots may
375/// not expose the source's live journal mode, so the owned marker is their
376/// authoritative readiness boundary.
377pub fn require_snapshot_initialized<E>(
378    connection: &Connection,
379    schema: SchemaVersion,
380) -> Result<(), InitializationError<E>> {
381    if schema_is_ready(connection, schema)? {
382        Ok(())
383    } else {
384        Err(InitializationError::SchemaNotInitialized {
385            name: schema.name,
386            version: schema.version,
387        })
388    }
389}
390
391fn require_file_initialized_impl<E>(
392    connection: &Connection,
393    busy_timeout: Duration,
394    schema: SchemaVersion,
395    on_readiness_contention: impl FnOnce(),
396) -> Result<(), InitializationError<E>> {
397    configure_busy_timeout(connection, busy_timeout)?;
398    if fast_path_is_ready(connection, schema)? {
399        return Ok(());
400    }
401
402    let _readiness_lock =
403        acquire_readiness_lock(connection, schema, lock_timeout(), on_readiness_contention)?;
404    if is_wal_journal_mode(connection)? && schema_is_ready(connection, schema)? {
405        return Ok(());
406    }
407    Err(InitializationError::SchemaNotInitialized {
408        name: schema.name,
409        version: schema.version,
410    })
411}
412
413/// Initialize a transient database with the same atomic schema-marker contract.
414///
415/// No filesystem lock or WAL promotion is performed because the connection is
416/// not shared across processes. A process-local lock serializes connection
417/// configuration, readiness inspection, and first use of named shared-memory
418/// databases because shared-cache schema reads conflict with schema creation.
419/// The callback follows the same transaction-local effects and at-most-once
420/// invocation contract as [`initialize_file`].
421pub fn initialize_transient<E, F>(
422    connection: &Connection,
423    busy_timeout: Duration,
424    schema: SchemaVersion,
425    initialize: F,
426) -> Result<(), InitializationError<E>>
427where
428    F: FnOnce(&Transaction<'_>) -> Result<(), E>,
429{
430    configure_busy_timeout(connection, busy_timeout)?;
431    if let Some(path) = main_database_path(connection) {
432        return Err(InitializationError::FileBackedTransient { path });
433    }
434    let _initialization_lock = TRANSIENT_INITIALIZATION_LOCK
435        .lock()
436        .unwrap_or_else(std::sync::PoisonError::into_inner);
437    configure_connection(connection)?;
438    if schema_is_ready(connection, schema)? {
439        return Ok(());
440    }
441    initialize_schema(connection, schema, initialize)
442}
443
444fn initialize_schema<E, F>(
445    connection: &Connection,
446    schema: SchemaVersion,
447    initialize: F,
448) -> Result<(), InitializationError<E>>
449where
450    F: FnOnce(&Transaction<'_>) -> Result<(), E>,
451{
452    let transaction = Transaction::new_unchecked(connection, TransactionBehavior::Immediate)
453        .map_err(InitializationError::Transaction)?;
454    transaction
455        .execute_batch(CREATE_SCHEMA_MARKER_TABLE)
456        .map_err(InitializationError::SchemaMarker)?;
457    if schema_marker_is_ready(&transaction, schema)? {
458        return transaction.commit().map_err(InitializationError::Commit);
459    }
460    initialize(&transaction).map_err(InitializationError::Initialize)?;
461    transaction
462        .execute(
463            "INSERT INTO main._harn_sqlite_schema_versions(name, version) VALUES (?1, ?2)
464             ON CONFLICT(name) DO UPDATE SET version = excluded.version",
465            params![schema.name, schema.version],
466        )
467        .map_err(InitializationError::SchemaMarker)?;
468    transaction.commit().map_err(InitializationError::Commit)
469}
470
471fn configure_busy_timeout<E>(
472    connection: &Connection,
473    busy_timeout: Duration,
474) -> Result<(), InitializationError<E>> {
475    let milliseconds = busy_timeout.as_millis();
476    if milliseconds > i32::MAX as u128 {
477        return Err(InitializationError::BusyTimeoutTooLarge { milliseconds });
478    }
479    connection
480        .busy_timeout(busy_timeout)
481        .map_err(InitializationError::BusyTimeout)
482}
483
484fn configure_connection<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
485    connection
486        .pragma_update(None, "synchronous", "NORMAL")
487        .map_err(InitializationError::Synchronous)
488}
489
490/// Environment override for [`lock_timeout`], in whole seconds.
491pub const LOCK_TIMEOUT_SECONDS_ENV: &str = "HARN_SQLITE_LOCK_TIMEOUT_SECONDS";
492
493/// Default wait for the sidecar lock.
494///
495/// The critical section it guards is WAL promotion plus one schema
496/// transaction — milliseconds of work, and only on the first open of a
497/// database. Three orders of magnitude of headroom means expiry says the
498/// holder is wedged, not that the disk was slow.
499const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
500
501/// How long to wait for the sidecar lock.
502///
503/// Operators can raise this for a filesystem where the default is genuinely
504/// too tight. An unparseable or zero value falls back to the default rather
505/// than failing the open: this is a diagnostic bound, and a typo in it must not
506/// take a database offline.
507fn lock_timeout() -> Duration {
508    std::env::var(LOCK_TIMEOUT_SECONDS_ENV)
509        .ok()
510        .and_then(|value| value.trim().parse::<u64>().ok())
511        .filter(|seconds| *seconds > 0)
512        .map_or(DEFAULT_LOCK_TIMEOUT, Duration::from_secs)
513}
514
515fn acquire_initialization_lock<E>(
516    connection: &Connection,
517    timeout: Duration,
518) -> Result<SqliteInitializationLock, InitializationError<E>> {
519    let path = initialization_lock_path(connection)?;
520    let file = OpenOptions::new()
521        .create(true)
522        .truncate(false)
523        .read(true)
524        .write(true)
525        .open(&path)
526        .map_err(|source| InitializationError::InitializationLockOpen {
527            path: path.clone(),
528            source,
529        })?;
530    harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Exclusive, timeout)
531        .map_err(InitializationError::InitializationLock)?;
532    Ok(SqliteInitializationLock { file })
533}
534
535fn acquire_readiness_lock<E>(
536    connection: &Connection,
537    schema: SchemaVersion,
538    timeout: Duration,
539    on_contention: impl FnOnce(),
540) -> Result<SqliteInitializationLock, InitializationError<E>> {
541    let path = initialization_lock_path(connection)?;
542    let file = match OpenOptions::new().read(true).open(&path) {
543        Ok(file) => file,
544        Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
545            return Err(InitializationError::SchemaNotInitialized {
546                name: schema.name,
547                version: schema.version,
548            });
549        }
550        Err(source) => {
551            return Err(InitializationError::InitializationLockOpen { path, source });
552        }
553    };
554    match file.try_lock_shared() {
555        Ok(()) => {}
556        Err(TryLockError::WouldBlock) => {
557            on_contention();
558            harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Shared, timeout)
559                .map_err(InitializationError::InitializationLock)?;
560        }
561        Err(TryLockError::Error(source)) => {
562            return Err(InitializationError::InitializationLockAcquire { path, source });
563        }
564    }
565    Ok(SqliteInitializationLock { file })
566}
567
568fn initialization_lock_path<E>(connection: &Connection) -> Result<PathBuf, InitializationError<E>> {
569    let database_path =
570        main_database_path(connection).ok_or(InitializationError::DatabasePathUnavailable)?;
571    let canonical = std::fs::canonicalize(&database_path).map_err(|source| {
572        InitializationError::DatabasePath {
573            path: database_path.clone(),
574            source,
575        }
576    })?;
577    let mut path = OsString::from(canonical.as_os_str());
578    path.push(".harn-init.lock");
579    Ok(PathBuf::from(path))
580}
581
582#[cfg(unix)]
583fn main_database_path(connection: &Connection) -> Option<PathBuf> {
584    use std::ffi::{CStr, OsStr};
585    use std::os::unix::ffi::OsStrExt;
586
587    // sqlite3_db_filename owns this pointer for the connection lifetime. Copy
588    // its bytes immediately so Unix paths retain their exact VFS identity.
589    let filename = unsafe {
590        let pointer =
591            rusqlite::ffi::sqlite3_db_filename(connection.handle(), rusqlite::MAIN_DB.as_ptr());
592        (!pointer.is_null()).then(|| CStr::from_ptr(pointer).to_bytes())
593    }?;
594    (!filename.is_empty()).then(|| PathBuf::from(OsStr::from_bytes(filename)))
595}
596
597#[cfg(not(unix))]
598fn main_database_path(connection: &Connection) -> Option<PathBuf> {
599    connection
600        .path()
601        .filter(|path| !path.is_empty())
602        .map(PathBuf::from)
603}
604
605struct SqliteInitializationLock {
606    file: File,
607}
608
609impl Drop for SqliteInitializationLock {
610    fn drop(&mut self) {
611        // Keep the lock file linked so queued openers cannot split across
612        // different inodes. Process exit also releases the advisory lock.
613        let _ = self.file.unlock();
614    }
615}
616
617fn schema_is_ready<E>(
618    connection: &Connection,
619    schema: SchemaVersion,
620) -> Result<bool, InitializationError<E>> {
621    let marker_exists = connection
622        .query_row(
623            "SELECT EXISTS(
624                SELECT 1 FROM main.sqlite_schema WHERE type = 'table' AND name = ?1
625             )",
626            params![SCHEMA_MARKER_TABLE],
627            |row| row.get::<_, bool>(0),
628        )
629        .map_err(InitializationError::SchemaReadiness)?;
630    if !marker_exists {
631        return Ok(false);
632    }
633    schema_marker_is_ready(connection, schema)
634}
635
636fn schema_marker_is_ready<E>(
637    connection: &Connection,
638    schema: SchemaVersion,
639) -> Result<bool, InitializationError<E>> {
640    let stored = connection
641        .query_row(
642            "SELECT version FROM main._harn_sqlite_schema_versions WHERE name = ?1",
643            params![schema.name],
644            |row| row.get::<_, i64>(0),
645        )
646        .optional()
647        .map_err(InitializationError::SchemaReadiness)?;
648    match stored {
649        Some(version) if version > schema.version => Err(InitializationError::NewerSchemaVersion {
650            name: schema.name,
651            stored: version,
652            supported: schema.version,
653        }),
654        Some(version) => Ok(version == schema.version),
655        None => Ok(false),
656    }
657}
658
659fn is_wal_journal_mode<E>(connection: &Connection) -> Result<bool, InitializationError<E>> {
660    current_journal_mode(connection)
661        .map(|mode| mode.eq_ignore_ascii_case("wal"))
662        .map_err(InitializationError::JournalModeQuery)
663}
664
665fn ensure_wal_journal_mode<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
666    if is_wal_journal_mode(connection)? {
667        return Ok(());
668    }
669    match connection.query_row("PRAGMA journal_mode = WAL", [], |row| {
670        row.get::<_, String>(0)
671    }) {
672        Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
673        Ok(mode) => Err(InitializationError::WalNotEnabled { mode }),
674        Err(error) if is_sqlite_busy_or_locked(&error) => match current_journal_mode(connection) {
675            Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
676            Ok(mode) => Err(InitializationError::WalBusyNotWal { mode }),
677            Err(query_error) => Err(InitializationError::WalBusyQuery {
678                wal_error: Box::new(error),
679                query_error: Box::new(query_error),
680            }),
681        },
682        Err(error) => Err(InitializationError::WalPragma(error)),
683    }
684}
685
686fn current_journal_mode(connection: &Connection) -> Result<String, rusqlite::Error> {
687    connection.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))
688}
689
690fn is_sqlite_busy_or_locked(error: &rusqlite::Error) -> bool {
691    sqlite_contention(error).is_some()
692}
693
694#[cfg(test)]
695#[path = "tests.rs"]
696mod tests;
harn_sqlite/lib.rs

harn_sqlite/
lib.rs
