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::preflight_schema_eligibility(&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 initialize_current: preflight_base,
414 allowed_existing_versions: &[1],
415 released_predecessors: &[],
416 owned_objects: &[crate::ledger::SchemaObject {
417 kind: crate::ledger::SchemaObjectKind::Table,
418 name: "preflight_t",
419 }],
420 retired_objects: &[],
421 };
422
423 #[test]
424 fn schema_preflight_refuses_future_file_before_wal_conversion() {
425 let dir = tempfile::tempdir().expect("tempdir");
426 let path = dir.path().join("db.sqlite3");
427 {
428 let conn = Connection::open(&path).expect("create raw");
435 conn.execute_batch(
436 "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
437 INSERT INTO meerkat_schema VALUES ('preflight-domain', 999);",
438 )
439 .expect("seed future ledger");
440 }
441 let err = open_with(
442 &path,
443 ConnectionProfile::PRIMARY,
444 OpenOptions {
445 schema_preflight: &[&PREFLIGHT_DOMAIN],
446 ..OpenOptions::default()
447 },
448 )
449 .expect_err("future file must be refused");
450 assert!(matches!(
451 err,
452 SqliteStoreError::SchemaFromTheFuture {
453 found: 999,
454 supported: 1,
455 ..
456 }
457 ));
458 let conn = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY)
461 .expect("reopen raw");
462 let journal: String = conn
463 .pragma_query_value(None, "journal_mode", |r| r.get(0))
464 .expect("journal_mode");
465 assert_eq!(journal, "delete");
466 }
467
468 #[test]
469 fn schema_preflight_allows_fresh_create_and_current_files() {
470 let dir = tempfile::tempdir().expect("tempdir");
471 let path = dir.path().join("fresh.sqlite3");
472 let options = OpenOptions {
473 schema_preflight: &[&PREFLIGHT_DOMAIN],
474 ..OpenOptions::default()
475 };
476 let mut conn = open_with(&path, ConnectionProfile::PRIMARY, options)
479 .expect("fresh create carries no ledger table and passes preflight");
480 crate::ledger::apply_domain_migrations(&mut conn, &PREFLIGHT_DOMAIN).expect("stamp");
481 drop(conn);
482 open_with(&path, ConnectionProfile::PRIMARY, options)
483 .expect("current file passes preflight");
484 }
485
486 #[test]
487 fn schema_preflight_refuses_current_fingerprint_mismatch_before_wal_conversion() {
488 let dir = tempfile::tempdir().expect("tempdir");
489 let path = dir.path().join("partial.sqlite3");
490 {
491 let conn = Connection::open(&path).expect("create raw");
492 conn.execute_batch(
493 "CREATE TABLE preflight_t (x INTEGER, candidate_only TEXT);
494 CREATE TABLE meerkat_schema (
495 domain TEXT PRIMARY KEY,
496 version INTEGER NOT NULL
497 );
498 INSERT INTO meerkat_schema VALUES ('preflight-domain', 1);",
499 )
500 .expect("seed partial current");
501 }
502 let err = open_with(
503 &path,
504 ConnectionProfile::PRIMARY,
505 OpenOptions {
506 schema_preflight: &[&PREFLIGHT_DOMAIN],
507 ..OpenOptions::default()
508 },
509 )
510 .expect_err("partial current must be refused");
511 assert!(matches!(
512 err,
513 SqliteStoreError::SchemaFingerprintMismatch { version: 1, .. }
514 ));
515 let conn = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY)
516 .expect("reopen raw");
517 let journal: String = conn
518 .pragma_query_value(None, "journal_mode", |row| row.get(0))
519 .expect("journal mode");
520 assert_eq!(journal, "delete");
521 }
522
523 #[test]
524 fn primary_in_memory_reports_memory_journal_and_opens() {
525 let conn = open(Path::new(":memory:"), ConnectionProfile::PRIMARY).expect("open");
528 let journal: String = conn
529 .pragma_query_value(None, "journal_mode", |r| r.get(0))
530 .expect("journal_mode");
531 assert_eq!(journal, "memory");
532 }
533
534 #[test]
535 fn write_contact_names_the_wal_sidecar_caveat() {
536 assert_eq!(
537 ConnectionProfile::PRIMARY.write_contact(),
538 WriteContact::ReadWrite
539 );
540 assert_eq!(
541 ConnectionProfile::Primary { create: false }.write_contact(),
542 WriteContact::ReadWrite
543 );
544 assert_eq!(
545 ConnectionProfile::Maintenance { write: true }.write_contact(),
546 WriteContact::ReadWrite
547 );
548 assert_eq!(
549 ConnectionProfile::ReadOnly.write_contact(),
550 WriteContact::ReadOnlyWalSidecars
551 );
552 assert_eq!(
553 ConnectionProfile::Maintenance { write: false }.write_contact(),
554 WriteContact::ReadOnlyWalSidecars
555 );
556 }
557}