1use std::path::Path;
28use std::time::Duration;
29
30use rusqlite::{Connection, OpenFlags, Transaction, TransactionBehavior};
31
32use crate::error::SqliteStoreError;
33use crate::ledger::SchemaDomain;
34
35pub const SHARED_BUSY_TIMEOUT: Duration = Duration::from_millis(60_000);
44
45const MAX_BUSY_TIMEOUT: Duration = Duration::from_millis(i32::MAX as u64);
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum ConnectionProfile {
56 Primary { create: bool },
61 ReadOnly,
67 Maintenance { write: bool },
71}
72
73#[derive(Debug, Clone, Copy, PartialEq, Eq)]
77pub enum WriteContact {
78 ReadWrite,
81 ReadOnlyWalSidecars,
90}
91
92impl ConnectionProfile {
93 pub const PRIMARY: Self = Self::Primary { create: true };
95
96 fn name(self) -> &'static str {
97 match self {
98 Self::Primary { create: true } => "primary",
99 Self::Primary { create: false } => "primary(no-create)",
100 Self::ReadOnly => "read-only",
101 Self::Maintenance { write: true } => "maintenance(write)",
102 Self::Maintenance { write: false } => "maintenance(read)",
103 }
104 }
105
106 fn default_busy_timeout(self) -> Duration {
107 match self {
108 Self::Primary { .. } | Self::ReadOnly => SHARED_BUSY_TIMEOUT,
109 Self::Maintenance { .. } => Duration::ZERO,
110 }
111 }
112
113 pub fn write_contact(self) -> WriteContact {
115 match self {
116 Self::Primary { .. } | Self::Maintenance { write: true } => WriteContact::ReadWrite,
117 Self::ReadOnly | Self::Maintenance { write: false } => {
118 WriteContact::ReadOnlyWalSidecars
119 }
120 }
121 }
122}
123
124#[derive(Debug, Clone, Copy, Default)]
127pub struct OpenOptions {
128 pub busy_timeout: Option<Duration>,
134 pub schema_preflight: &'static [&'static SchemaDomain],
149}
150
151pub fn open(path: &Path, profile: ConnectionProfile) -> Result<Connection, SqliteStoreError> {
153 open_with(path, profile, OpenOptions::default())
154}
155
156pub fn open_with(
158 path: &Path,
159 profile: ConnectionProfile,
160 options: OpenOptions,
161) -> Result<Connection, SqliteStoreError> {
162 let conn = match profile {
163 ConnectionProfile::Primary { create: true } => {
164 if let Some(parent) = path.parent() {
165 std::fs::create_dir_all(parent)?;
166 }
167 Connection::open(path)?
168 }
169 ConnectionProfile::Primary { create: false } => {
170 open_existing(path, profile, OpenFlags::SQLITE_OPEN_READ_WRITE)?
171 }
172 ConnectionProfile::ReadOnly => {
173 open_existing(path, profile, OpenFlags::SQLITE_OPEN_READ_ONLY)?
174 }
175 ConnectionProfile::Maintenance { write } => {
176 let base = if write {
177 OpenFlags::SQLITE_OPEN_READ_WRITE
178 } else {
179 OpenFlags::SQLITE_OPEN_READ_ONLY
180 };
181 open_existing(path, profile, base)?
182 }
183 };
184
185 let busy = options
186 .busy_timeout
187 .unwrap_or_else(|| profile.default_busy_timeout())
188 .min(MAX_BUSY_TIMEOUT);
189 conn.busy_timeout(busy)?;
190
191 for domain in options.schema_preflight {
199 crate::ledger::refuse_future_schema(&conn, domain)?;
200 }
201
202 if let ConnectionProfile::Primary { .. } = profile {
203 set_wal_journal_mode(&conn, path, busy)?;
204 conn.pragma_update(None, "synchronous", "FULL")?;
205 }
206
207 Ok(conn)
208}
209
210fn set_wal_journal_mode(
219 conn: &Connection,
220 path: &Path,
221 busy_timeout: Duration,
222) -> Result<(), SqliteStoreError> {
223 let deadline = std::time::Instant::now() + busy_timeout.max(Duration::from_millis(250));
226 loop {
227 let result = conn
233 .pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get::<_, String>(0));
234 match result {
235 Ok(mode) if mode.eq_ignore_ascii_case("wal") || mode.eq_ignore_ascii_case("memory") => {
236 return Ok(());
237 }
238 Ok(mode) => {
239 return Err(SqliteStoreError::WalNotEstablished {
240 path: path.to_path_buf(),
241 actual: mode,
242 });
243 }
244 Err(error)
245 if crate::error::is_busy_or_locked(&error)
246 && std::time::Instant::now() < deadline =>
247 {
248 std::thread::sleep(Duration::from_millis(5));
249 }
250 Err(error) => return Err(error.into()),
251 }
252 }
253}
254
255fn open_existing(
256 path: &Path,
257 profile: ConnectionProfile,
258 base: OpenFlags,
259) -> Result<Connection, SqliteStoreError> {
260 if !path.is_file() {
261 return Err(SqliteStoreError::OpenRefused {
262 path: path.to_path_buf(),
263 profile: profile.name(),
264 detail: "database file does not exist".to_string(),
265 });
266 }
267 Ok(Connection::open_with_flags(
268 path,
269 base | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX,
270 )?)
271}
272
273pub fn begin_immediate(conn: &mut Connection) -> Result<Transaction<'_>, SqliteStoreError> {
277 Ok(conn.transaction_with_behavior(TransactionBehavior::Immediate)?)
278}
279
280#[cfg(test)]
281#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
282mod tests {
283 use super::*;
284
285 #[test]
286 fn primary_creates_dirs_sets_wal_and_full() {
287 let dir = tempfile::tempdir().expect("tempdir");
288 let path = dir.path().join("nested/sub/test.sqlite3");
289 let conn = open(&path, ConnectionProfile::PRIMARY).expect("open");
290 let journal: String = conn
291 .pragma_query_value(None, "journal_mode", |r| r.get(0))
292 .expect("journal_mode");
293 assert_eq!(journal, "wal");
294 let sync: i64 = conn
295 .pragma_query_value(None, "synchronous", |r| r.get(0))
296 .expect("synchronous");
297 assert_eq!(sync, 2, "synchronous=FULL");
298 }
299
300 #[test]
301 fn non_creating_profiles_refuse_missing_file() {
302 let dir = tempfile::tempdir().expect("tempdir");
303 let path = dir.path().join("missing.sqlite3");
304 for profile in [
305 ConnectionProfile::Primary { create: false },
306 ConnectionProfile::ReadOnly,
307 ConnectionProfile::Maintenance { write: true },
308 ConnectionProfile::Maintenance { write: false },
309 ] {
310 let err = open(&path, profile).expect_err("must refuse missing file");
311 assert!(
312 matches!(err, SqliteStoreError::OpenRefused { .. }),
313 "{profile:?}"
314 );
315 }
316 assert!(
317 !path.exists(),
318 "no profile may create the file as a side effect"
319 );
320 }
321
322 #[test]
323 fn read_only_profile_rejects_writes_and_preserves_journal_mode() {
324 let dir = tempfile::tempdir().expect("tempdir");
325 let path = dir.path().join("db.sqlite3");
326 {
327 let conn = Connection::open(&path).expect("create");
328 conn.execute_batch("CREATE TABLE t (x INTEGER)")
329 .expect("ddl");
330 }
331 let conn = open(&path, ConnectionProfile::ReadOnly).expect("open ro");
332 let journal: String = conn
333 .pragma_query_value(None, "journal_mode", |r| r.get(0))
334 .expect("journal_mode");
335 assert_eq!(
336 journal, "delete",
337 "reader must not convert the journal mode"
338 );
339 conn.execute("INSERT INTO t VALUES (1)", [])
340 .expect_err("read-only connection must reject writes");
341 }
342
343 #[test]
344 fn maintenance_profile_fails_fast_on_held_lock() {
345 let dir = tempfile::tempdir().expect("tempdir");
346 let path = dir.path().join("db.sqlite3");
347 let mut writer = open(&path, ConnectionProfile::PRIMARY).expect("writer");
348 writer
349 .execute_batch("CREATE TABLE t (x INTEGER)")
350 .expect("ddl");
351 let tx = begin_immediate(&mut writer).expect("hold write lock");
352
353 let maint = open(&path, ConnectionProfile::Maintenance { write: true }).expect("open");
354 let err = maint
355 .execute_batch("BEGIN IMMEDIATE")
356 .expect_err("zero busy timeout must surface the held lock immediately");
357 assert!(crate::error::is_busy_or_locked(&match err {
358 rusqlite::Error::SqliteFailure(..) => err,
359 other => panic!("unexpected error shape: {other}"),
360 }));
361 drop(tx);
362 }
363
364 #[test]
365 fn busy_timeout_override_applies() {
366 let dir = tempfile::tempdir().expect("tempdir");
367 let path = dir.path().join("db.sqlite3");
368 let conn = open_with(
369 &path,
370 ConnectionProfile::PRIMARY,
371 OpenOptions {
372 busy_timeout: Some(Duration::from_millis(1234)),
373 ..OpenOptions::default()
374 },
375 )
376 .expect("open");
377 let timeout: i64 = conn
378 .pragma_query_value(None, "busy_timeout", |r| r.get(0))
379 .expect("busy_timeout");
380 assert_eq!(timeout, 1234);
381 }
382
383 #[test]
384 fn oversized_busy_timeout_is_clamped_not_panicking() {
385 let dir = tempfile::tempdir().expect("tempdir");
386 let path = dir.path().join("db.sqlite3");
387 let conn = open_with(
388 &path,
389 ConnectionProfile::PRIMARY,
390 OpenOptions {
391 busy_timeout: Some(Duration::MAX),
392 ..OpenOptions::default()
393 },
394 )
395 .expect("oversized timeout must clamp, not panic");
396 let timeout: i64 = conn
397 .pragma_query_value(None, "busy_timeout", |r| r.get(0))
398 .expect("busy_timeout");
399 assert_eq!(timeout, i64::from(i32::MAX));
400 }
401
402 fn preflight_base(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
403 tx.execute_batch("CREATE TABLE IF NOT EXISTS preflight_t (x INTEGER)")
404 }
405
406 const PREFLIGHT_DOMAIN: SchemaDomain = SchemaDomain {
407 name: "preflight-domain",
408 migrations: &[crate::ledger::Migration {
409 version: 1,
410 name: "base",
411 apply: preflight_base,
412 }],
413 };
414
415 #[test]
416 fn schema_preflight_refuses_future_file_before_wal_conversion() {
417 let dir = tempfile::tempdir().expect("tempdir");
418 let path = dir.path().join("db.sqlite3");
419 {
420 let conn = Connection::open(&path).expect("create raw");
427 conn.execute_batch(
428 "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
429 INSERT INTO meerkat_schema VALUES ('preflight-domain', 999);",
430 )
431 .expect("seed future ledger");
432 }
433 let err = open_with(
434 &path,
435 ConnectionProfile::PRIMARY,
436 OpenOptions {
437 schema_preflight: &[&PREFLIGHT_DOMAIN],
438 ..OpenOptions::default()
439 },
440 )
441 .expect_err("future file must be refused");
442 assert!(matches!(
443 err,
444 SqliteStoreError::SchemaFromTheFuture {
445 found: 999,
446 supported: 1,
447 ..
448 }
449 ));
450 let conn = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY)
453 .expect("reopen raw");
454 let journal: String = conn
455 .pragma_query_value(None, "journal_mode", |r| r.get(0))
456 .expect("journal_mode");
457 assert_eq!(journal, "delete");
458 }
459
460 #[test]
461 fn schema_preflight_allows_fresh_create_and_current_files() {
462 let dir = tempfile::tempdir().expect("tempdir");
463 let path = dir.path().join("fresh.sqlite3");
464 let options = OpenOptions {
465 schema_preflight: &[&PREFLIGHT_DOMAIN],
466 ..OpenOptions::default()
467 };
468 let mut conn = open_with(&path, ConnectionProfile::PRIMARY, options)
471 .expect("fresh create carries no ledger table and passes preflight");
472 crate::ledger::apply_domain_migrations(&mut conn, &PREFLIGHT_DOMAIN).expect("stamp");
473 drop(conn);
474 open_with(&path, ConnectionProfile::PRIMARY, options)
475 .expect("current file passes preflight");
476 }
477
478 #[test]
479 fn primary_in_memory_reports_memory_journal_and_opens() {
480 let conn = open(Path::new(":memory:"), ConnectionProfile::PRIMARY).expect("open");
483 let journal: String = conn
484 .pragma_query_value(None, "journal_mode", |r| r.get(0))
485 .expect("journal_mode");
486 assert_eq!(journal, "memory");
487 }
488
489 #[test]
490 fn write_contact_names_the_wal_sidecar_caveat() {
491 assert_eq!(
492 ConnectionProfile::PRIMARY.write_contact(),
493 WriteContact::ReadWrite
494 );
495 assert_eq!(
496 ConnectionProfile::Primary { create: false }.write_contact(),
497 WriteContact::ReadWrite
498 );
499 assert_eq!(
500 ConnectionProfile::Maintenance { write: true }.write_contact(),
501 WriteContact::ReadWrite
502 );
503 assert_eq!(
504 ConnectionProfile::ReadOnly.write_contact(),
505 WriteContact::ReadOnlyWalSidecars
506 );
507 assert_eq!(
508 ConnectionProfile::Maintenance { write: false }.write_contact(),
509 WriteContact::ReadOnlyWalSidecars
510 );
511 }
512}