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
18const SCHEMA_MARKER_TABLE: &str = "_harn_sqlite_schema_versions";
19const CREATE_SCHEMA_MARKER_TABLE: &str =
20 "CREATE TABLE IF NOT EXISTS main._harn_sqlite_schema_versions (
21 name TEXT PRIMARY KEY,
22 version INTEGER NOT NULL CHECK(version > 0)
23);";
24static TRANSIENT_INITIALIZATION_LOCK: Mutex<()> = Mutex::new(());
25
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
28pub struct SchemaVersion {
29 name: &'static str,
30 version: i64,
31}
32
33#[derive(Clone, Copy, Debug, PartialEq, Eq)]
35pub enum SqliteContention {
36 Busy,
38 Locked,
40}
41
42#[must_use]
44pub fn sqlite_contention(error: &rusqlite::Error) -> Option<SqliteContention> {
45 match error {
46 rusqlite::Error::SqliteFailure(failure, _) => match failure.code {
47 ErrorCode::DatabaseBusy => Some(SqliteContention::Busy),
48 ErrorCode::DatabaseLocked => Some(SqliteContention::Locked),
49 _ => None,
50 },
51 _ => None,
52 }
53}
54
55impl SchemaVersion {
56 #[must_use]
60 pub const fn new(name: &'static str, version: i64) -> Self {
61 assert!(!name.is_empty(), "SQLite schema name must not be empty");
62 assert!(version > 0, "SQLite schema version must be positive");
63 Self { name, version }
64 }
65}
66
67#[derive(Debug)]
69#[non_exhaustive]
70pub enum InitializationError<E> {
71 BusyTimeoutTooLarge { milliseconds: u128 },
73 BusyTimeout(rusqlite::Error),
75 JournalModeQuery(rusqlite::Error),
77 DatabasePathUnavailable,
79 FileBackedTransient { path: PathBuf },
81 DatabasePath {
83 path: PathBuf,
84 source: std::io::Error,
85 },
86 InitializationLockOpen {
88 path: PathBuf,
89 source: std::io::Error,
90 },
91 InitializationLockAcquire {
93 path: PathBuf,
94 source: std::io::Error,
95 },
96 InitializationLock(harn_flock::LockError),
99 WalPragma(rusqlite::Error),
101 WalNotEnabled { mode: String },
103 WalBusyNotWal { mode: String },
105 WalBusyQuery {
107 wal_error: Box<rusqlite::Error>,
108 query_error: Box<rusqlite::Error>,
109 },
110 Synchronous(rusqlite::Error),
112 SchemaReadiness(rusqlite::Error),
114 SchemaNotInitialized { name: &'static str, version: i64 },
117 NewerSchemaVersion {
119 name: &'static str,
120 stored: i64,
121 supported: i64,
122 },
123 Transaction(rusqlite::Error),
125 Initialize(E),
127 SchemaMarker(rusqlite::Error),
129 Commit(rusqlite::Error),
131}
132
133impl InitializationError<rusqlite::Error> {
134 #[must_use]
136 pub fn is_busy_or_locked(&self) -> bool {
137 if let Self::Initialize(error) = self {
138 is_sqlite_busy_or_locked(error)
139 } else {
140 initialization_stage_is_busy_or_locked(self)
141 }
142 }
143}
144
145impl<E: fmt::Display> fmt::Display for InitializationError<E> {
146 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
147 match self {
148 Self::BusyTimeoutTooLarge { milliseconds } => write!(
149 f,
150 "busy_timeout {milliseconds}ms exceeds SQLite's maximum"
151 ),
152 Self::BusyTimeout(error) => write!(f, "busy_timeout failed: {error}"),
153 Self::JournalModeQuery(error) => write!(f, "journal_mode query failed: {error}"),
154 Self::DatabasePathUnavailable => {
155 write!(f, "SQLite connection has no file-backed main database path")
156 }
157 Self::FileBackedTransient { path } => write!(
158 f,
159 "transient SQLite initialization requires a private non-file database, got {}",
160 path.display()
161 ),
162 Self::DatabasePath { path, source } => write!(
163 f,
164 "could not resolve SQLite database path {}: {source}",
165 path.display()
166 ),
167 Self::InitializationLockOpen { path, source } => write!(
168 f,
169 "could not open SQLite initialization lock {}: {source}",
170 path.display()
171 ),
172 Self::InitializationLockAcquire { path, source } => write!(
173 f,
174 "could not acquire SQLite initialization lock {}: {source}",
175 path.display()
176 ),
177 Self::InitializationLock(error) => {
178 write!(f, "SQLite initialization lock unavailable: {error}")
179 }
180 Self::WalPragma(error) => write!(f, "WAL journal_mode pragma failed: {error}"),
181 Self::WalNotEnabled { mode } => {
182 write!(f, "WAL journal_mode request returned {mode}")
183 }
184 Self::WalBusyNotWal { mode } => {
185 write!(f, "WAL journal_mode request left journal_mode at {mode}")
186 }
187 Self::WalBusyQuery {
188 wal_error,
189 query_error,
190 } => write!(
191 f,
192 "WAL journal_mode pragma failed: {wal_error}; journal_mode query also failed: {query_error}"
193 ),
194 Self::Synchronous(error) => write!(f, "synchronous pragma failed: {error}"),
195 Self::SchemaReadiness(error) => write!(f, "schema readiness query failed: {error}"),
196 Self::SchemaNotInitialized { name, version } => {
197 write!(f, "SQLite schema {name} version {version} is not initialized")
198 }
199 Self::NewerSchemaVersion {
200 name,
201 stored,
202 supported,
203 } => write!(
204 f,
205 "SQLite schema {name} version {stored} is newer than supported version {supported}"
206 ),
207 Self::Transaction(error) => write!(f, "schema transaction failed: {error}"),
208 Self::Initialize(error) => write!(f, "schema initialization failed: {error}"),
209 Self::SchemaMarker(error) => write!(f, "schema marker update failed: {error}"),
210 Self::Commit(error) => write!(f, "schema transaction commit failed: {error}"),
211 }
212 }
213}
214
215impl<E: std::error::Error + 'static> std::error::Error for InitializationError<E> {
216 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
217 match self {
218 Self::BusyTimeout(error)
219 | Self::JournalModeQuery(error)
220 | Self::WalPragma(error)
221 | Self::Synchronous(error)
222 | Self::SchemaReadiness(error)
223 | Self::Transaction(error)
224 | Self::SchemaMarker(error)
225 | Self::Commit(error) => Some(error),
226 Self::DatabasePath { source, .. }
227 | Self::InitializationLockOpen { source, .. }
228 | Self::InitializationLockAcquire { source, .. } => Some(source),
229 Self::InitializationLock(error) => Some(error),
230 Self::WalBusyQuery { wal_error, .. } => Some(wal_error),
231 Self::Initialize(error) => Some(error),
232 Self::BusyTimeoutTooLarge { .. }
233 | Self::DatabasePathUnavailable
234 | Self::FileBackedTransient { .. }
235 | Self::SchemaNotInitialized { .. }
236 | Self::WalNotEnabled { .. }
237 | Self::WalBusyNotWal { .. }
238 | Self::NewerSchemaVersion { .. } => None,
239 }
240 }
241}
242
243pub fn initialize_file<E, F>(
254 connection: &Connection,
255 busy_timeout: Duration,
256 schema: SchemaVersion,
257 initialize: F,
258) -> Result<(), InitializationError<E>>
259where
260 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
261{
262 configure_busy_timeout(connection, busy_timeout)?;
263 if fast_path_is_ready(connection, schema)? {
264 return configure_connection(connection);
265 }
266
267 let _initialization_lock = acquire_initialization_lock(connection, lock_timeout())?;
268 ensure_wal_journal_mode(connection)?;
269 configure_connection(connection)?;
270 initialize_schema(connection, schema, initialize)
271}
272
273fn fast_path_is_ready<E>(
274 connection: &Connection,
275 schema: SchemaVersion,
276) -> Result<bool, InitializationError<E>> {
277 match is_wal_journal_mode(connection) {
278 Ok(true) => {}
279 Ok(false) => return Ok(false),
280 Err(error) if initialization_stage_is_busy_or_locked(&error) => return Ok(false),
281 Err(error) => return Err(error),
282 }
283 match schema_is_ready(connection, schema) {
284 Ok(ready) => Ok(ready),
285 Err(error) if initialization_stage_is_busy_or_locked(&error) => Ok(false),
286 Err(error) => Err(error),
287 }
288}
289
290fn initialization_stage_is_busy_or_locked<E>(error: &InitializationError<E>) -> bool {
291 match error {
292 InitializationError::BusyTimeout(error)
293 | InitializationError::JournalModeQuery(error)
294 | InitializationError::WalPragma(error)
295 | InitializationError::Synchronous(error)
296 | InitializationError::SchemaReadiness(error)
297 | InitializationError::Transaction(error)
298 | InitializationError::SchemaMarker(error)
299 | InitializationError::Commit(error) => is_sqlite_busy_or_locked(error),
300 InitializationError::WalBusyNotWal { .. } => true,
301 InitializationError::WalBusyQuery {
302 wal_error,
303 query_error,
304 } => is_sqlite_busy_or_locked(wal_error) || is_sqlite_busy_or_locked(query_error),
305 InitializationError::BusyTimeoutTooLarge { .. }
306 | InitializationError::DatabasePath { .. }
307 | InitializationError::DatabasePathUnavailable
308 | InitializationError::FileBackedTransient { .. }
309 | InitializationError::InitializationLockOpen { .. }
310 | InitializationError::InitializationLockAcquire { .. }
311 | InitializationError::InitializationLock(_)
312 | InitializationError::SchemaNotInitialized { .. }
313 | InitializationError::NewerSchemaVersion { .. }
314 | InitializationError::WalNotEnabled { .. }
315 | InitializationError::Initialize(_) => false,
316 }
317}
318
319pub fn require_file_initialized<E>(
328 connection: &Connection,
329 busy_timeout: Duration,
330 schema: SchemaVersion,
331) -> Result<(), InitializationError<E>> {
332 require_file_initialized_impl(connection, busy_timeout, schema, || {})
333}
334
335pub fn require_snapshot_initialized<E>(
344 connection: &Connection,
345 schema: SchemaVersion,
346) -> Result<(), InitializationError<E>> {
347 if schema_is_ready(connection, schema)? {
348 Ok(())
349 } else {
350 Err(InitializationError::SchemaNotInitialized {
351 name: schema.name,
352 version: schema.version,
353 })
354 }
355}
356
357fn require_file_initialized_impl<E>(
358 connection: &Connection,
359 busy_timeout: Duration,
360 schema: SchemaVersion,
361 on_readiness_contention: impl FnOnce(),
362) -> Result<(), InitializationError<E>> {
363 configure_busy_timeout(connection, busy_timeout)?;
364 if fast_path_is_ready(connection, schema)? {
365 return Ok(());
366 }
367
368 let _readiness_lock =
369 acquire_readiness_lock(connection, schema, lock_timeout(), on_readiness_contention)?;
370 if is_wal_journal_mode(connection)? && schema_is_ready(connection, schema)? {
371 return Ok(());
372 }
373 Err(InitializationError::SchemaNotInitialized {
374 name: schema.name,
375 version: schema.version,
376 })
377}
378
379pub fn initialize_transient<E, F>(
388 connection: &Connection,
389 busy_timeout: Duration,
390 schema: SchemaVersion,
391 initialize: F,
392) -> Result<(), InitializationError<E>>
393where
394 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
395{
396 configure_busy_timeout(connection, busy_timeout)?;
397 if let Some(path) = main_database_path(connection) {
398 return Err(InitializationError::FileBackedTransient { path });
399 }
400 let _initialization_lock = TRANSIENT_INITIALIZATION_LOCK
401 .lock()
402 .unwrap_or_else(std::sync::PoisonError::into_inner);
403 configure_connection(connection)?;
404 if schema_is_ready(connection, schema)? {
405 return Ok(());
406 }
407 initialize_schema(connection, schema, initialize)
408}
409
410fn initialize_schema<E, F>(
411 connection: &Connection,
412 schema: SchemaVersion,
413 initialize: F,
414) -> Result<(), InitializationError<E>>
415where
416 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
417{
418 let transaction = Transaction::new_unchecked(connection, TransactionBehavior::Immediate)
419 .map_err(InitializationError::Transaction)?;
420 transaction
421 .execute_batch(CREATE_SCHEMA_MARKER_TABLE)
422 .map_err(InitializationError::SchemaMarker)?;
423 if schema_marker_is_ready(&transaction, schema)? {
424 return transaction.commit().map_err(InitializationError::Commit);
425 }
426 initialize(&transaction).map_err(InitializationError::Initialize)?;
427 transaction
428 .execute(
429 "INSERT INTO main._harn_sqlite_schema_versions(name, version) VALUES (?1, ?2)
430 ON CONFLICT(name) DO UPDATE SET version = excluded.version",
431 params![schema.name, schema.version],
432 )
433 .map_err(InitializationError::SchemaMarker)?;
434 transaction.commit().map_err(InitializationError::Commit)
435}
436
437fn configure_busy_timeout<E>(
438 connection: &Connection,
439 busy_timeout: Duration,
440) -> Result<(), InitializationError<E>> {
441 let milliseconds = busy_timeout.as_millis();
442 if milliseconds > i32::MAX as u128 {
443 return Err(InitializationError::BusyTimeoutTooLarge { milliseconds });
444 }
445 connection
446 .busy_timeout(busy_timeout)
447 .map_err(InitializationError::BusyTimeout)
448}
449
450fn configure_connection<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
451 connection
452 .pragma_update(None, "synchronous", "NORMAL")
453 .map_err(InitializationError::Synchronous)
454}
455
456pub const LOCK_TIMEOUT_SECONDS_ENV: &str = "HARN_SQLITE_LOCK_TIMEOUT_SECONDS";
458
459const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
466
467fn lock_timeout() -> Duration {
474 std::env::var(LOCK_TIMEOUT_SECONDS_ENV)
475 .ok()
476 .and_then(|value| value.trim().parse::<u64>().ok())
477 .filter(|seconds| *seconds > 0)
478 .map_or(DEFAULT_LOCK_TIMEOUT, Duration::from_secs)
479}
480
481fn acquire_initialization_lock<E>(
482 connection: &Connection,
483 timeout: Duration,
484) -> Result<SqliteInitializationLock, InitializationError<E>> {
485 let path = initialization_lock_path(connection)?;
486 let file = OpenOptions::new()
487 .create(true)
488 .truncate(false)
489 .read(true)
490 .write(true)
491 .open(&path)
492 .map_err(|source| InitializationError::InitializationLockOpen {
493 path: path.clone(),
494 source,
495 })?;
496 harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Exclusive, timeout)
497 .map_err(InitializationError::InitializationLock)?;
498 Ok(SqliteInitializationLock { file })
499}
500
501fn acquire_readiness_lock<E>(
502 connection: &Connection,
503 schema: SchemaVersion,
504 timeout: Duration,
505 on_contention: impl FnOnce(),
506) -> Result<SqliteInitializationLock, InitializationError<E>> {
507 let path = initialization_lock_path(connection)?;
508 let file = match OpenOptions::new().read(true).open(&path) {
509 Ok(file) => file,
510 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
511 return Err(InitializationError::SchemaNotInitialized {
512 name: schema.name,
513 version: schema.version,
514 });
515 }
516 Err(source) => {
517 return Err(InitializationError::InitializationLockOpen { path, source });
518 }
519 };
520 match file.try_lock_shared() {
521 Ok(()) => {}
522 Err(TryLockError::WouldBlock) => {
523 on_contention();
524 harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Shared, timeout)
525 .map_err(InitializationError::InitializationLock)?;
526 }
527 Err(TryLockError::Error(source)) => {
528 return Err(InitializationError::InitializationLockAcquire { path, source });
529 }
530 }
531 Ok(SqliteInitializationLock { file })
532}
533
534fn initialization_lock_path<E>(connection: &Connection) -> Result<PathBuf, InitializationError<E>> {
535 let database_path =
536 main_database_path(connection).ok_or(InitializationError::DatabasePathUnavailable)?;
537 let canonical = std::fs::canonicalize(&database_path).map_err(|source| {
538 InitializationError::DatabasePath {
539 path: database_path.clone(),
540 source,
541 }
542 })?;
543 let mut path = OsString::from(canonical.as_os_str());
544 path.push(".harn-init.lock");
545 Ok(PathBuf::from(path))
546}
547
548#[cfg(unix)]
549fn main_database_path(connection: &Connection) -> Option<PathBuf> {
550 use std::ffi::{CStr, OsStr};
551 use std::os::unix::ffi::OsStrExt;
552
553 let filename = unsafe {
556 let pointer =
557 rusqlite::ffi::sqlite3_db_filename(connection.handle(), rusqlite::MAIN_DB.as_ptr());
558 (!pointer.is_null()).then(|| CStr::from_ptr(pointer).to_bytes())
559 }?;
560 (!filename.is_empty()).then(|| PathBuf::from(OsStr::from_bytes(filename)))
561}
562
563#[cfg(not(unix))]
564fn main_database_path(connection: &Connection) -> Option<PathBuf> {
565 connection
566 .path()
567 .filter(|path| !path.is_empty())
568 .map(PathBuf::from)
569}
570
571struct SqliteInitializationLock {
572 file: File,
573}
574
575impl Drop for SqliteInitializationLock {
576 fn drop(&mut self) {
577 let _ = self.file.unlock();
580 }
581}
582
583fn schema_is_ready<E>(
584 connection: &Connection,
585 schema: SchemaVersion,
586) -> Result<bool, InitializationError<E>> {
587 let marker_exists = connection
588 .query_row(
589 "SELECT EXISTS(
590 SELECT 1 FROM main.sqlite_schema WHERE type = 'table' AND name = ?1
591 )",
592 params![SCHEMA_MARKER_TABLE],
593 |row| row.get::<_, bool>(0),
594 )
595 .map_err(InitializationError::SchemaReadiness)?;
596 if !marker_exists {
597 return Ok(false);
598 }
599 schema_marker_is_ready(connection, schema)
600}
601
602fn schema_marker_is_ready<E>(
603 connection: &Connection,
604 schema: SchemaVersion,
605) -> Result<bool, InitializationError<E>> {
606 let stored = connection
607 .query_row(
608 "SELECT version FROM main._harn_sqlite_schema_versions WHERE name = ?1",
609 params![schema.name],
610 |row| row.get::<_, i64>(0),
611 )
612 .optional()
613 .map_err(InitializationError::SchemaReadiness)?;
614 match stored {
615 Some(version) if version > schema.version => Err(InitializationError::NewerSchemaVersion {
616 name: schema.name,
617 stored: version,
618 supported: schema.version,
619 }),
620 Some(version) => Ok(version == schema.version),
621 None => Ok(false),
622 }
623}
624
625fn is_wal_journal_mode<E>(connection: &Connection) -> Result<bool, InitializationError<E>> {
626 current_journal_mode(connection)
627 .map(|mode| mode.eq_ignore_ascii_case("wal"))
628 .map_err(InitializationError::JournalModeQuery)
629}
630
631fn ensure_wal_journal_mode<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
632 if is_wal_journal_mode(connection)? {
633 return Ok(());
634 }
635 match connection.query_row("PRAGMA journal_mode = WAL", [], |row| {
636 row.get::<_, String>(0)
637 }) {
638 Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
639 Ok(mode) => Err(InitializationError::WalNotEnabled { mode }),
640 Err(error) if is_sqlite_busy_or_locked(&error) => match current_journal_mode(connection) {
641 Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
642 Ok(mode) => Err(InitializationError::WalBusyNotWal { mode }),
643 Err(query_error) => Err(InitializationError::WalBusyQuery {
644 wal_error: Box::new(error),
645 query_error: Box::new(query_error),
646 }),
647 },
648 Err(error) => Err(InitializationError::WalPragma(error)),
649 }
650}
651
652fn current_journal_mode(connection: &Connection) -> Result<String, rusqlite::Error> {
653 connection.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))
654}
655
656fn is_sqlite_busy_or_locked(error: &rusqlite::Error) -> bool {
657 sqlite_contention(error).is_some()
658}
659
660#[cfg(test)]
661#[path = "tests.rs"]
662mod tests;