use std::path::Path;
use std::time::Duration;
use rusqlite::{Connection, OpenFlags, Transaction, TransactionBehavior};
use crate::error::SqliteStoreError;
use crate::ledger::SchemaDomain;
pub const SHARED_BUSY_TIMEOUT: Duration = Duration::from_millis(60_000);
const MAX_BUSY_TIMEOUT: Duration = Duration::from_millis(i32::MAX as u64);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionProfile {
Primary { create: bool },
ReadOnly,
Maintenance { write: bool },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WriteContact {
ReadWrite,
ReadOnlyWalSidecars,
}
impl ConnectionProfile {
pub const PRIMARY: Self = Self::Primary { create: true };
fn name(self) -> &'static str {
match self {
Self::Primary { create: true } => "primary",
Self::Primary { create: false } => "primary(no-create)",
Self::ReadOnly => "read-only",
Self::Maintenance { write: true } => "maintenance(write)",
Self::Maintenance { write: false } => "maintenance(read)",
}
}
fn default_busy_timeout(self) -> Duration {
match self {
Self::Primary { .. } | Self::ReadOnly => SHARED_BUSY_TIMEOUT,
Self::Maintenance { .. } => Duration::ZERO,
}
}
pub fn write_contact(self) -> WriteContact {
match self {
Self::Primary { .. } | Self::Maintenance { write: true } => WriteContact::ReadWrite,
Self::ReadOnly | Self::Maintenance { write: false } => {
WriteContact::ReadOnlyWalSidecars
}
}
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct OpenOptions {
pub busy_timeout: Option<Duration>,
pub schema_preflight: &'static [&'static SchemaDomain],
}
pub fn open(path: &Path, profile: ConnectionProfile) -> Result<Connection, SqliteStoreError> {
open_with(path, profile, OpenOptions::default())
}
pub fn open_with(
path: &Path,
profile: ConnectionProfile,
options: OpenOptions,
) -> Result<Connection, SqliteStoreError> {
let conn = match profile {
ConnectionProfile::Primary { create: true } => {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
Connection::open(path)?
}
ConnectionProfile::Primary { create: false } => {
open_existing(path, profile, OpenFlags::SQLITE_OPEN_READ_WRITE)?
}
ConnectionProfile::ReadOnly => {
open_existing(path, profile, OpenFlags::SQLITE_OPEN_READ_ONLY)?
}
ConnectionProfile::Maintenance { write } => {
let base = if write {
OpenFlags::SQLITE_OPEN_READ_WRITE
} else {
OpenFlags::SQLITE_OPEN_READ_ONLY
};
open_existing(path, profile, base)?
}
};
let busy = options
.busy_timeout
.unwrap_or_else(|| profile.default_busy_timeout())
.min(MAX_BUSY_TIMEOUT);
conn.busy_timeout(busy)?;
for domain in options.schema_preflight {
crate::ledger::refuse_future_schema(&conn, domain)?;
}
if let ConnectionProfile::Primary { .. } = profile {
set_wal_journal_mode(&conn, path, busy)?;
conn.pragma_update(None, "synchronous", "FULL")?;
}
Ok(conn)
}
fn set_wal_journal_mode(
conn: &Connection,
path: &Path,
busy_timeout: Duration,
) -> Result<(), SqliteStoreError> {
let deadline = std::time::Instant::now() + busy_timeout.max(Duration::from_millis(250));
loop {
let result = conn
.pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get::<_, String>(0));
match result {
Ok(mode) if mode.eq_ignore_ascii_case("wal") || mode.eq_ignore_ascii_case("memory") => {
return Ok(());
}
Ok(mode) => {
return Err(SqliteStoreError::WalNotEstablished {
path: path.to_path_buf(),
actual: mode,
});
}
Err(error)
if crate::error::is_busy_or_locked(&error)
&& std::time::Instant::now() < deadline =>
{
std::thread::sleep(Duration::from_millis(5));
}
Err(error) => return Err(error.into()),
}
}
}
fn open_existing(
path: &Path,
profile: ConnectionProfile,
base: OpenFlags,
) -> Result<Connection, SqliteStoreError> {
if !path.is_file() {
return Err(SqliteStoreError::OpenRefused {
path: path.to_path_buf(),
profile: profile.name(),
detail: "database file does not exist".to_string(),
});
}
Ok(Connection::open_with_flags(
path,
base | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX,
)?)
}
pub fn begin_immediate(conn: &mut Connection) -> Result<Transaction<'_>, SqliteStoreError> {
Ok(conn.transaction_with_behavior(TransactionBehavior::Immediate)?)
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
use super::*;
#[test]
fn primary_creates_dirs_sets_wal_and_full() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("nested/sub/test.sqlite3");
let conn = open(&path, ConnectionProfile::PRIMARY).expect("open");
let journal: String = conn
.pragma_query_value(None, "journal_mode", |r| r.get(0))
.expect("journal_mode");
assert_eq!(journal, "wal");
let sync: i64 = conn
.pragma_query_value(None, "synchronous", |r| r.get(0))
.expect("synchronous");
assert_eq!(sync, 2, "synchronous=FULL");
}
#[test]
fn non_creating_profiles_refuse_missing_file() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("missing.sqlite3");
for profile in [
ConnectionProfile::Primary { create: false },
ConnectionProfile::ReadOnly,
ConnectionProfile::Maintenance { write: true },
ConnectionProfile::Maintenance { write: false },
] {
let err = open(&path, profile).expect_err("must refuse missing file");
assert!(
matches!(err, SqliteStoreError::OpenRefused { .. }),
"{profile:?}"
);
}
assert!(
!path.exists(),
"no profile may create the file as a side effect"
);
}
#[test]
fn read_only_profile_rejects_writes_and_preserves_journal_mode() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("db.sqlite3");
{
let conn = Connection::open(&path).expect("create");
conn.execute_batch("CREATE TABLE t (x INTEGER)")
.expect("ddl");
}
let conn = open(&path, ConnectionProfile::ReadOnly).expect("open ro");
let journal: String = conn
.pragma_query_value(None, "journal_mode", |r| r.get(0))
.expect("journal_mode");
assert_eq!(
journal, "delete",
"reader must not convert the journal mode"
);
conn.execute("INSERT INTO t VALUES (1)", [])
.expect_err("read-only connection must reject writes");
}
#[test]
fn maintenance_profile_fails_fast_on_held_lock() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("db.sqlite3");
let mut writer = open(&path, ConnectionProfile::PRIMARY).expect("writer");
writer
.execute_batch("CREATE TABLE t (x INTEGER)")
.expect("ddl");
let tx = begin_immediate(&mut writer).expect("hold write lock");
let maint = open(&path, ConnectionProfile::Maintenance { write: true }).expect("open");
let err = maint
.execute_batch("BEGIN IMMEDIATE")
.expect_err("zero busy timeout must surface the held lock immediately");
assert!(crate::error::is_busy_or_locked(&match err {
rusqlite::Error::SqliteFailure(..) => err,
other => panic!("unexpected error shape: {other}"),
}));
drop(tx);
}
#[test]
fn busy_timeout_override_applies() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("db.sqlite3");
let conn = open_with(
&path,
ConnectionProfile::PRIMARY,
OpenOptions {
busy_timeout: Some(Duration::from_millis(1234)),
..OpenOptions::default()
},
)
.expect("open");
let timeout: i64 = conn
.pragma_query_value(None, "busy_timeout", |r| r.get(0))
.expect("busy_timeout");
assert_eq!(timeout, 1234);
}
#[test]
fn oversized_busy_timeout_is_clamped_not_panicking() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("db.sqlite3");
let conn = open_with(
&path,
ConnectionProfile::PRIMARY,
OpenOptions {
busy_timeout: Some(Duration::MAX),
..OpenOptions::default()
},
)
.expect("oversized timeout must clamp, not panic");
let timeout: i64 = conn
.pragma_query_value(None, "busy_timeout", |r| r.get(0))
.expect("busy_timeout");
assert_eq!(timeout, i64::from(i32::MAX));
}
fn preflight_base(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE IF NOT EXISTS preflight_t (x INTEGER)")
}
const PREFLIGHT_DOMAIN: SchemaDomain = SchemaDomain {
name: "preflight-domain",
migrations: &[crate::ledger::Migration {
version: 1,
name: "base",
apply: preflight_base,
}],
};
#[test]
fn schema_preflight_refuses_future_file_before_wal_conversion() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("db.sqlite3");
{
let conn = Connection::open(&path).expect("create raw");
conn.execute_batch(
"CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
INSERT INTO meerkat_schema VALUES ('preflight-domain', 999);",
)
.expect("seed future ledger");
}
let err = open_with(
&path,
ConnectionProfile::PRIMARY,
OpenOptions {
schema_preflight: &[&PREFLIGHT_DOMAIN],
..OpenOptions::default()
},
)
.expect_err("future file must be refused");
assert!(matches!(
err,
SqliteStoreError::SchemaFromTheFuture {
found: 999,
supported: 1,
..
}
));
let conn = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY)
.expect("reopen raw");
let journal: String = conn
.pragma_query_value(None, "journal_mode", |r| r.get(0))
.expect("journal_mode");
assert_eq!(journal, "delete");
}
#[test]
fn schema_preflight_allows_fresh_create_and_current_files() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("fresh.sqlite3");
let options = OpenOptions {
schema_preflight: &[&PREFLIGHT_DOMAIN],
..OpenOptions::default()
};
let mut conn = open_with(&path, ConnectionProfile::PRIMARY, options)
.expect("fresh create carries no ledger table and passes preflight");
crate::ledger::apply_domain_migrations(&mut conn, &PREFLIGHT_DOMAIN).expect("stamp");
drop(conn);
open_with(&path, ConnectionProfile::PRIMARY, options)
.expect("current file passes preflight");
}
#[test]
fn primary_in_memory_reports_memory_journal_and_opens() {
let conn = open(Path::new(":memory:"), ConnectionProfile::PRIMARY).expect("open");
let journal: String = conn
.pragma_query_value(None, "journal_mode", |r| r.get(0))
.expect("journal_mode");
assert_eq!(journal, "memory");
}
#[test]
fn write_contact_names_the_wal_sidecar_caveat() {
assert_eq!(
ConnectionProfile::PRIMARY.write_contact(),
WriteContact::ReadWrite
);
assert_eq!(
ConnectionProfile::Primary { create: false }.write_contact(),
WriteContact::ReadWrite
);
assert_eq!(
ConnectionProfile::Maintenance { write: true }.write_contact(),
WriteContact::ReadWrite
);
assert_eq!(
ConnectionProfile::ReadOnly.write_contact(),
WriteContact::ReadOnlyWalSidecars
);
assert_eq!(
ConnectionProfile::Maintenance { write: false }.write_contact(),
WriteContact::ReadOnlyWalSidecars
);
}
}