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
335fn require_file_initialized_impl<E>(
336 connection: &Connection,
337 busy_timeout: Duration,
338 schema: SchemaVersion,
339 on_readiness_contention: impl FnOnce(),
340) -> Result<(), InitializationError<E>> {
341 configure_busy_timeout(connection, busy_timeout)?;
342 if fast_path_is_ready(connection, schema)? {
343 return Ok(());
344 }
345
346 let _readiness_lock =
347 acquire_readiness_lock(connection, schema, lock_timeout(), on_readiness_contention)?;
348 if is_wal_journal_mode(connection)? && schema_is_ready(connection, schema)? {
349 return Ok(());
350 }
351 Err(InitializationError::SchemaNotInitialized {
352 name: schema.name,
353 version: schema.version,
354 })
355}
356
357pub fn initialize_transient<E, F>(
366 connection: &Connection,
367 busy_timeout: Duration,
368 schema: SchemaVersion,
369 initialize: F,
370) -> Result<(), InitializationError<E>>
371where
372 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
373{
374 configure_busy_timeout(connection, busy_timeout)?;
375 if let Some(path) = main_database_path(connection) {
376 return Err(InitializationError::FileBackedTransient { path });
377 }
378 let _initialization_lock = TRANSIENT_INITIALIZATION_LOCK
379 .lock()
380 .unwrap_or_else(std::sync::PoisonError::into_inner);
381 configure_connection(connection)?;
382 if schema_is_ready(connection, schema)? {
383 return Ok(());
384 }
385 initialize_schema(connection, schema, initialize)
386}
387
388fn initialize_schema<E, F>(
389 connection: &Connection,
390 schema: SchemaVersion,
391 initialize: F,
392) -> Result<(), InitializationError<E>>
393where
394 F: FnOnce(&Transaction<'_>) -> Result<(), E>,
395{
396 let transaction = Transaction::new_unchecked(connection, TransactionBehavior::Immediate)
397 .map_err(InitializationError::Transaction)?;
398 transaction
399 .execute_batch(CREATE_SCHEMA_MARKER_TABLE)
400 .map_err(InitializationError::SchemaMarker)?;
401 if schema_marker_is_ready(&transaction, schema)? {
402 return transaction.commit().map_err(InitializationError::Commit);
403 }
404 initialize(&transaction).map_err(InitializationError::Initialize)?;
405 transaction
406 .execute(
407 "INSERT INTO main._harn_sqlite_schema_versions(name, version) VALUES (?1, ?2)
408 ON CONFLICT(name) DO UPDATE SET version = excluded.version",
409 params![schema.name, schema.version],
410 )
411 .map_err(InitializationError::SchemaMarker)?;
412 transaction.commit().map_err(InitializationError::Commit)
413}
414
415fn configure_busy_timeout<E>(
416 connection: &Connection,
417 busy_timeout: Duration,
418) -> Result<(), InitializationError<E>> {
419 let milliseconds = busy_timeout.as_millis();
420 if milliseconds > i32::MAX as u128 {
421 return Err(InitializationError::BusyTimeoutTooLarge { milliseconds });
422 }
423 connection
424 .busy_timeout(busy_timeout)
425 .map_err(InitializationError::BusyTimeout)
426}
427
428fn configure_connection<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
429 connection
430 .pragma_update(None, "synchronous", "NORMAL")
431 .map_err(InitializationError::Synchronous)
432}
433
434pub const LOCK_TIMEOUT_SECONDS_ENV: &str = "HARN_SQLITE_LOCK_TIMEOUT_SECONDS";
436
437const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
444
445fn lock_timeout() -> Duration {
452 std::env::var(LOCK_TIMEOUT_SECONDS_ENV)
453 .ok()
454 .and_then(|value| value.trim().parse::<u64>().ok())
455 .filter(|seconds| *seconds > 0)
456 .map_or(DEFAULT_LOCK_TIMEOUT, Duration::from_secs)
457}
458
459fn acquire_initialization_lock<E>(
460 connection: &Connection,
461 timeout: Duration,
462) -> Result<SqliteInitializationLock, InitializationError<E>> {
463 let path = initialization_lock_path(connection)?;
464 let file = OpenOptions::new()
465 .create(true)
466 .truncate(false)
467 .read(true)
468 .write(true)
469 .open(&path)
470 .map_err(|source| InitializationError::InitializationLockOpen {
471 path: path.clone(),
472 source,
473 })?;
474 harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Exclusive, timeout)
475 .map_err(InitializationError::InitializationLock)?;
476 Ok(SqliteInitializationLock { file })
477}
478
479fn acquire_readiness_lock<E>(
480 connection: &Connection,
481 schema: SchemaVersion,
482 timeout: Duration,
483 on_contention: impl FnOnce(),
484) -> Result<SqliteInitializationLock, InitializationError<E>> {
485 let path = initialization_lock_path(connection)?;
486 let file = match OpenOptions::new().read(true).open(&path) {
487 Ok(file) => file,
488 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
489 return Err(InitializationError::SchemaNotInitialized {
490 name: schema.name,
491 version: schema.version,
492 });
493 }
494 Err(source) => {
495 return Err(InitializationError::InitializationLockOpen { path, source });
496 }
497 };
498 match file.try_lock_shared() {
499 Ok(()) => {}
500 Err(TryLockError::WouldBlock) => {
501 on_contention();
502 harn_flock::lock_with_deadline(&file, &path, harn_flock::LockMode::Shared, timeout)
503 .map_err(InitializationError::InitializationLock)?;
504 }
505 Err(TryLockError::Error(source)) => {
506 return Err(InitializationError::InitializationLockAcquire { path, source });
507 }
508 }
509 Ok(SqliteInitializationLock { file })
510}
511
512fn initialization_lock_path<E>(connection: &Connection) -> Result<PathBuf, InitializationError<E>> {
513 let database_path =
514 main_database_path(connection).ok_or(InitializationError::DatabasePathUnavailable)?;
515 let canonical = std::fs::canonicalize(&database_path).map_err(|source| {
516 InitializationError::DatabasePath {
517 path: database_path.clone(),
518 source,
519 }
520 })?;
521 let mut path = OsString::from(canonical.as_os_str());
522 path.push(".harn-init.lock");
523 Ok(PathBuf::from(path))
524}
525
526#[cfg(unix)]
527fn main_database_path(connection: &Connection) -> Option<PathBuf> {
528 use std::ffi::{CStr, OsStr};
529 use std::os::unix::ffi::OsStrExt;
530
531 let filename = unsafe {
534 let pointer =
535 rusqlite::ffi::sqlite3_db_filename(connection.handle(), rusqlite::MAIN_DB.as_ptr());
536 (!pointer.is_null()).then(|| CStr::from_ptr(pointer).to_bytes())
537 }?;
538 (!filename.is_empty()).then(|| PathBuf::from(OsStr::from_bytes(filename)))
539}
540
541#[cfg(not(unix))]
542fn main_database_path(connection: &Connection) -> Option<PathBuf> {
543 connection
544 .path()
545 .filter(|path| !path.is_empty())
546 .map(PathBuf::from)
547}
548
549struct SqliteInitializationLock {
550 file: File,
551}
552
553impl Drop for SqliteInitializationLock {
554 fn drop(&mut self) {
555 let _ = self.file.unlock();
558 }
559}
560
561fn schema_is_ready<E>(
562 connection: &Connection,
563 schema: SchemaVersion,
564) -> Result<bool, InitializationError<E>> {
565 let marker_exists = connection
566 .query_row(
567 "SELECT EXISTS(
568 SELECT 1 FROM main.sqlite_schema WHERE type = 'table' AND name = ?1
569 )",
570 params![SCHEMA_MARKER_TABLE],
571 |row| row.get::<_, bool>(0),
572 )
573 .map_err(InitializationError::SchemaReadiness)?;
574 if !marker_exists {
575 return Ok(false);
576 }
577 schema_marker_is_ready(connection, schema)
578}
579
580fn schema_marker_is_ready<E>(
581 connection: &Connection,
582 schema: SchemaVersion,
583) -> Result<bool, InitializationError<E>> {
584 let stored = connection
585 .query_row(
586 "SELECT version FROM main._harn_sqlite_schema_versions WHERE name = ?1",
587 params![schema.name],
588 |row| row.get::<_, i64>(0),
589 )
590 .optional()
591 .map_err(InitializationError::SchemaReadiness)?;
592 match stored {
593 Some(version) if version > schema.version => Err(InitializationError::NewerSchemaVersion {
594 name: schema.name,
595 stored: version,
596 supported: schema.version,
597 }),
598 Some(version) => Ok(version == schema.version),
599 None => Ok(false),
600 }
601}
602
603fn is_wal_journal_mode<E>(connection: &Connection) -> Result<bool, InitializationError<E>> {
604 current_journal_mode(connection)
605 .map(|mode| mode.eq_ignore_ascii_case("wal"))
606 .map_err(InitializationError::JournalModeQuery)
607}
608
609fn ensure_wal_journal_mode<E>(connection: &Connection) -> Result<(), InitializationError<E>> {
610 if is_wal_journal_mode(connection)? {
611 return Ok(());
612 }
613 match connection.query_row("PRAGMA journal_mode = WAL", [], |row| {
614 row.get::<_, String>(0)
615 }) {
616 Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
617 Ok(mode) => Err(InitializationError::WalNotEnabled { mode }),
618 Err(error) if is_sqlite_busy_or_locked(&error) => match current_journal_mode(connection) {
619 Ok(mode) if mode.eq_ignore_ascii_case("wal") => Ok(()),
620 Ok(mode) => Err(InitializationError::WalBusyNotWal { mode }),
621 Err(query_error) => Err(InitializationError::WalBusyQuery {
622 wal_error: Box::new(error),
623 query_error: Box::new(query_error),
624 }),
625 },
626 Err(error) => Err(InitializationError::WalPragma(error)),
627 }
628}
629
630fn current_journal_mode(connection: &Connection) -> Result<String, rusqlite::Error> {
631 connection.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))
632}
633
634fn is_sqlite_busy_or_locked(error: &rusqlite::Error) -> bool {
635 sqlite_contention(error).is_some()
636}
637
638#[cfg(test)]
639#[path = "tests.rs"]
640mod tests;