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 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/// Stable identity for one schema stored in a Harn-owned SQLite database.
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub struct SchemaVersion {
32    name: &'static str,
33    version: i64,
34}
35
36/// Lock-contention reason reported by SQLite.
37#[derive(Clone, Copy, Debug, PartialEq, Eq)]
38pub enum SqliteContention {
39    /// Another connection currently owns the database write lock.
40    Busy,
41    /// A shared-cache table lock prevents the operation from proceeding.
42    Locked,
43}
44
45/// Classify a SQLite error without relying on rendered error text.
46#[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    /// Define a non-empty schema name and positive version.
60    ///
61    /// Invalid constants fail during compile-time evaluation.
62    #[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/// Failure to configure or initialize a Harn-owned SQLite database.
71#[derive(Debug)]
72#[non_exhaustive]
73pub enum InitializationError<E> {
74    /// The configured busy timeout cannot be represented by SQLite.
75    BusyTimeoutTooLarge { milliseconds: u128 },
76    /// The connection rejected its configured busy timeout.
77    BusyTimeout(rusqlite::Error),
78    /// The current journal mode could not be observed.
79    JournalModeQuery(rusqlite::Error),
80    /// The connection does not own a file-backed main database.
81    DatabasePathUnavailable,
82    /// A file-backed connection was passed to the transient initializer.
83    FileBackedTransient { path: PathBuf },
84    /// The opened database path could not be resolved to one lock identity.
85    DatabasePath {
86        path: PathBuf,
87        source: std::io::Error,
88    },
89    /// The persistent sidecar lock could not be opened.
90    InitializationLockOpen {
91        path: PathBuf,
92        source: std::io::Error,
93    },
94    /// Exclusive ownership of the sidecar lock could not be acquired.
95    InitializationLockAcquire {
96        path: PathBuf,
97        source: std::io::Error,
98    },
99    /// The sidecar lock was still held when the wait budget expired, or the
100    /// operating system refused it. Carries the path and the elapsed wait.
101    InitializationLock(harn_flock::LockError),
102    /// SQLite rejected WAL journal mode for a non-contention reason.
103    WalPragma(rusqlite::Error),
104    /// SQLite accepted the journal-mode request but returned another mode.
105    WalNotEnabled { mode: String },
106    /// WAL promotion was busy and the database remained in another mode.
107    WalBusyNotWal { mode: String },
108    /// WAL promotion and the diagnostic journal-mode query both failed.
109    WalBusyQuery {
110        wal_error: Box<rusqlite::Error>,
111        query_error: Box<rusqlite::Error>,
112    },
113    /// The connection rejected the runtime synchronous setting.
114    Synchronous(rusqlite::Error),
115    /// The schema marker could not be inspected.
116    SchemaReadiness(rusqlite::Error),
117    /// No initializer committed the exact schema version before the readiness
118    /// lease became available.
119    SchemaNotInitialized { name: &'static str, version: i64 },
120    /// The database was initialized by a newer incompatible schema owner.
121    NewerSchemaVersion {
122        name: &'static str,
123        stored: i64,
124        supported: i64,
125    },
126    /// The atomic schema transaction could not begin.
127    Transaction(rusqlite::Error),
128    /// The owning schema callback failed.
129    Initialize(E),
130    /// The schema marker table or row could not be written.
131    SchemaMarker(rusqlite::Error),
132    /// The schema and marker transaction could not commit.
133    Commit(rusqlite::Error),
134}
135
136impl InitializationError<rusqlite::Error> {
137    /// Whether the SQLite portion of this failure reports lock contention.
138    #[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    /// Stable failure stage, without database paths, SQL, or stored values.
248    #[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
276/// Initialize a file-backed Harn database exactly once per schema version.
277///
278/// The lock identity comes from the connection's VFS-resolved main-database
279/// filename, so callers cannot accidentally serialize a different path.
280/// Initialization serializes WAL promotion and the schema transaction across
281/// processes. The callback and version marker commit atomically. Ready opens
282/// observe WAL plus the exact marker and do not acquire the sidecar lock. The
283/// callback is invoked at most once by this function and should keep all
284/// persistent effects inside the supplied transaction so an error rolls them
285/// back with the marker.
286pub 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
352/// Require an exact schema version on a file-backed connection without
353/// initializing it.
354///
355/// Ready databases avoid the sidecar lock. When initialization is in flight,
356/// this waits for its persistent lease and validates the marker after the
357/// writer releases it. If no initializer owns the lease, an absent marker is
358/// reported here instead of allowing a read-only consumer to fail later on a
359/// missing application table.
360pub 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
368/// Require the exact schema marker on an inert SQLite snapshot.
369///
370/// Unlike [`require_file_initialized`], this does not inspect journal mode,
371/// configure the connection, or acquire the initialization lock. Callers must
372/// first isolate the source from SQLite's durable bookkeeping (for example,
373/// by opening an `immutable=1` URI or a disposable copy). Such snapshots may
374/// not expose the source's live journal mode, so the owned marker is their
375/// authoritative readiness boundary.
376pub 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
412/// Initialize a transient database with the same atomic schema-marker contract.
413///
414/// No filesystem lock or WAL promotion is performed because the connection is
415/// not shared across processes. A process-local lock serializes connection
416/// configuration, readiness inspection, and first use of named shared-memory
417/// databases because shared-cache schema reads conflict with schema creation.
418/// The callback follows the same transaction-local effects and at-most-once
419/// invocation contract as [`initialize_file`].
420pub 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
489/// Environment override for [`lock_timeout`], in whole seconds.
490pub const LOCK_TIMEOUT_SECONDS_ENV: &str = "HARN_SQLITE_LOCK_TIMEOUT_SECONDS";
491
492/// Default wait for the sidecar lock.
493///
494/// The critical section it guards is WAL promotion plus one schema
495/// transaction — milliseconds of work, and only on the first open of a
496/// database. Three orders of magnitude of headroom means expiry says the
497/// holder is wedged, not that the disk was slow.
498const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
499
500/// How long to wait for the sidecar lock.
501///
502/// Operators can raise this for a filesystem where the default is genuinely
503/// too tight. An unparseable or zero value falls back to the default rather
504/// than failing the open: this is a diagnostic bound, and a typo in it must not
505/// take a database offline.
506fn 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    // sqlite3_db_filename owns this pointer for the connection lifetime. Copy
587    // its bytes immediately so Unix paths retain their exact VFS identity.
588    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        // Keep the lock file linked so queued openers cannot split across
611        // different inodes. Process exit also releases the advisory lock.
612        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;
harn_sqlite/lib.rs

harn_sqlite/
lib.rs
