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

harn_sqlite/
lib.rs
