lib

Core libraries for Radroots
git clone https://radroots.dev/git/lib.git
Log | Files | Refs | README

commit 8332b86a7d3a9cbc2210038a1e9a21fca6f73312
parent 62ef402ac1af945b2daf1deccd1c87a1ce455282
Author: triesap <tyson@radroots.org>
Date:   Tue, 11 Aug 2026 11:59:26 +0000

service-sqlite: apply connection policy

- seal reviewed busy-timeout and bounded-pool configuration
- enforce WAL, FULL sync, foreign keys, trusted-schema, and query-only policy
- retain writer and inspection authority across private pooled connections
- reject path, inode, metadata, pragma, sidecar, and checkout drift

Diffstat:
MCargo.lock | 2++
Mcrates/service_sqlite/Cargo.toml | 2++
Mcrates/service_sqlite/src/authority.rs | 57++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
Acrates/service_sqlite/src/config.rs | 191+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Mcrates/service_sqlite/src/lib.rs | 2++
Mcrates/service_sqlite/src/open.rs | 1201+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Mcrates/service_sqlite/tests/package_boundary.rs | 82+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------
7 files changed, 1524 insertions(+), 13 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock @@ -3914,7 +3914,9 @@ dependencies = [ "rustix 1.1.4", "serde", "serde_json", + "sqlx", "tempfile", + "tokio", ] [[package]] diff --git a/crates/service_sqlite/Cargo.toml b/crates/service_sqlite/Cargo.toml @@ -16,10 +16,12 @@ fs2 = { workspace = true } radroots_runtime_paths = { workspace = true } rustix = { workspace = true } serde = { workspace = true, features = ["derive", "std"] } +sqlx = { workspace = true, features = ["runtime-tokio", "sqlite-bundled"] } [dev-dependencies] serde_json = { workspace = true } tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt"] } [lints] workspace = true diff --git a/crates/service_sqlite/src/authority.rs b/crates/service_sqlite/src/authority.rs @@ -1,7 +1,7 @@ //! Lifetime authority for the sole writable service database owner. use core::fmt; -use std::{error::Error, fs::File}; +use std::{error::Error, fs::File, path::PathBuf}; use fs2::FileExt; @@ -20,6 +20,11 @@ use crate::{OpenMode, ServiceSqliteError, ServiceSqliteErrorKind, ServiceSqliteP /// ``` pub struct WriterAuthority { file: Option<File>, + #[allow( + dead_code, + reason = "Step 056 binds the authority for the private Step 061 pool host" + )] + database_path: PathBuf, #[cfg(any(target_os = "linux", target_os = "macos"))] directory: File, } @@ -50,6 +55,24 @@ impl WriterAuthority { &self.directory } + #[allow( + dead_code, + reason = "Step 056 binds the authority for the private Step 061 pool host" + )] + pub(crate) fn validate_for( + &self, + paths: &ServiceSqlitePaths, + ) -> Result<(), ServiceSqliteError> { + if !self.is_held() || self.database_path != paths.state_database() { + return Err(authority_error(WriterAuthorityCause::Mismatched)); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + validate_directory_binding(&self.directory, paths).map_err(authority_error)?; + + Ok(()) + } + /// Explicitly releases writer authority; subsequent calls are no-ops. pub fn release(&mut self) -> Result<(), ServiceSqliteError> { let Some(file) = self.file.as_ref() else { @@ -100,6 +123,11 @@ enum WriterAuthorityCause { LockWrongOwner, #[cfg(any(target_os = "linux", target_os = "macos"))] Contended, + #[allow( + dead_code, + reason = "Step 056 binds the authority for the private Step 061 pool host" + )] + Mismatched, UnlockFailed, } @@ -128,6 +156,7 @@ impl fmt::Display for WriterAuthorityCause { Self::LockWrongOwner => "SQLite writer lock has the wrong owner", #[cfg(any(target_os = "linux", target_os = "macos"))] Self::Contended => "another SQLite writer is active", + Self::Mismatched => "SQLite writer authority does not match this database", Self::UnlockFailed => "SQLite writer authority could not be released", }) } @@ -187,6 +216,7 @@ fn acquire_supported(paths: &ServiceSqlitePaths) -> Result<WriterAuthority, Writ match FileExt::try_lock_exclusive(&file) { Ok(()) => Ok(WriterAuthority { file: Some(file), + database_path: paths.state_database().to_path_buf(), directory, }), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { @@ -239,6 +269,31 @@ fn validate_lock( Ok(()) } +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn validate_directory_binding( + held_directory: &File, + paths: &ServiceSqlitePaths, +) -> Result<(), WriterAuthorityCause> { + use rustix::fs::{Mode, OFlags, fstat, open}; + + let current_directory = open( + paths + .state_database() + .parent() + .ok_or(WriterAuthorityCause::Mismatched)?, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(|_| WriterAuthorityCause::Mismatched)?; + let held = fstat(held_directory).map_err(|_| WriterAuthorityCause::Mismatched)?; + let current = fstat(&current_directory).map_err(|_| WriterAuthorityCause::Mismatched)?; + if held.st_dev == current.st_dev && held.st_ino == current.st_ino { + Ok(()) + } else { + Err(WriterAuthorityCause::Mismatched) + } +} + #[cfg(all(test, any(target_os = "linux", target_os = "macos")))] mod tests { use std::{ diff --git a/crates/service_sqlite/src/config.rs b/crates/service_sqlite/src/config.rs @@ -0,0 +1,191 @@ +//! Validated connection limits for one service-owned SQLite database. + +use core::{fmt, time::Duration}; +use std::error::Error; + +const MIN_BUSY_TIMEOUT: Duration = Duration::from_millis(1); +const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_BUSY_TIMEOUT: Duration = Duration::from_secs(60); +const MIN_CONNECTIONS: u32 = 1; +const DEFAULT_MAX_CONNECTIONS: u32 = 8; +const MAX_CONNECTIONS: u32 = 8; + +/// Validated limits applied to every connection in a service SQLite pool. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ServiceSqliteConnectionOptions { + busy_timeout: Duration, + max_connections: u32, +} + +impl ServiceSqliteConnectionOptions { + /// Returns the reviewed service defaults: five seconds and eight connections. + #[must_use] + pub const fn reviewed() -> Self { + Self { + busy_timeout: DEFAULT_BUSY_TIMEOUT, + max_connections: DEFAULT_MAX_CONNECTIONS, + } + } + + /// Validates an integral-millisecond busy timeout and bounded pool size. + pub fn new( + busy_timeout: Duration, + max_connections: u32, + ) -> Result<Self, ServiceSqliteConnectionOptionsError> { + let timeout_nanos = busy_timeout.as_nanos(); + if !timeout_nanos.is_multiple_of(1_000_000) { + return Err(ServiceSqliteConnectionOptionsError::BusyTimeoutNotMilliseconds); + } + if busy_timeout < MIN_BUSY_TIMEOUT { + return Err(ServiceSqliteConnectionOptionsError::BusyTimeoutTooSmall); + } + if busy_timeout > MAX_BUSY_TIMEOUT { + return Err(ServiceSqliteConnectionOptionsError::BusyTimeoutTooLarge); + } + if max_connections < MIN_CONNECTIONS { + return Err(ServiceSqliteConnectionOptionsError::PoolTooSmall); + } + if max_connections > MAX_CONNECTIONS { + return Err(ServiceSqliteConnectionOptionsError::PoolTooLarge); + } + Ok(Self { + busy_timeout, + max_connections, + }) + } + + /// Returns the exact busy timeout applied to SQLite and pool acquisition. + #[must_use] + pub const fn busy_timeout(self) -> Duration { + self.busy_timeout + } + + /// Returns the maximum number of connections in the pool. + #[must_use] + pub const fn max_connections(self) -> u32 { + self.max_connections + } + + #[allow( + dead_code, + reason = "Step 056 keeps pragma verification private until the Step 061 host boundary" + )] + pub(crate) fn busy_timeout_milliseconds(self) -> i64 { + i64::try_from(self.busy_timeout.as_millis()) + .expect("validated busy timeout always fits in i64 milliseconds") + } +} + +impl Default for ServiceSqliteConnectionOptions { + fn default() -> Self { + Self::reviewed() + } +} + +/// Path-free failure returned when SQLite connection limits are invalid. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ServiceSqliteConnectionOptionsError { + BusyTimeoutTooSmall, + BusyTimeoutTooLarge, + BusyTimeoutNotMilliseconds, + PoolTooSmall, + PoolTooLarge, +} + +impl fmt::Display for ServiceSqliteConnectionOptionsError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::BusyTimeoutTooSmall => "SQLite busy timeout must be at least one millisecond", + Self::BusyTimeoutTooLarge => "SQLite busy timeout must not exceed sixty seconds", + Self::BusyTimeoutNotMilliseconds => { + "SQLite busy timeout must be an integral number of milliseconds" + } + Self::PoolTooSmall => "SQLite connection pool must contain at least one connection", + Self::PoolTooLarge => "SQLite connection pool must not exceed eight connections", + }) + } +} + +impl Error for ServiceSqliteConnectionOptionsError {} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reviewed_defaults_and_complete_boundary_inventory_are_exact() { + let reviewed = ServiceSqliteConnectionOptions::reviewed(); + assert_eq!(reviewed.busy_timeout(), Duration::from_secs(5)); + assert_eq!(reviewed.max_connections(), 8); + assert_eq!(ServiceSqliteConnectionOptions::default(), reviewed); + + for timeout in [ + Duration::from_millis(1), + Duration::from_secs(5), + Duration::from_secs(60), + ] { + for max_connections in 1..=8 { + assert_eq!( + ServiceSqliteConnectionOptions::new(timeout, max_connections), + Ok(ServiceSqliteConnectionOptions { + busy_timeout: timeout, + max_connections, + }) + ); + } + } + } + + #[test] + fn zero_fractional_oversized_and_extreme_values_fail_closed() { + let vectors = [ + ( + Duration::ZERO, + 8, + ServiceSqliteConnectionOptionsError::BusyTimeoutTooSmall, + ), + ( + Duration::from_nanos(1), + 8, + ServiceSqliteConnectionOptionsError::BusyTimeoutNotMilliseconds, + ), + ( + Duration::from_micros(1_500), + 8, + ServiceSqliteConnectionOptionsError::BusyTimeoutNotMilliseconds, + ), + ( + Duration::from_millis(60_001), + 8, + ServiceSqliteConnectionOptionsError::BusyTimeoutTooLarge, + ), + ( + Duration::MAX, + 8, + ServiceSqliteConnectionOptionsError::BusyTimeoutNotMilliseconds, + ), + ( + Duration::from_secs(5), + 0, + ServiceSqliteConnectionOptionsError::PoolTooSmall, + ), + ( + Duration::from_secs(5), + 9, + ServiceSqliteConnectionOptionsError::PoolTooLarge, + ), + ( + Duration::from_secs(5), + u32::MAX, + ServiceSqliteConnectionOptionsError::PoolTooLarge, + ), + ]; + + for (timeout, max_connections, expected) in vectors { + let error = ServiceSqliteConnectionOptions::new(timeout, max_connections) + .expect_err("invalid options must fail"); + assert_eq!(error, expected); + assert!(!error.to_string().contains(&max_connections.to_string())); + } + } +} diff --git a/crates/service_sqlite/src/lib.rs b/crates/service_sqlite/src/lib.rs @@ -3,12 +3,14 @@ //! Reusable, service-neutral SQLite mechanics for Radroots services. mod authority; +mod config; mod error; mod initialize; mod open; mod status; pub use authority::WriterAuthority; +pub use config::{ServiceSqliteConnectionOptions, ServiceSqliteConnectionOptionsError}; pub use error::{ SafeServiceSqliteError, ServiceSqliteError, ServiceSqliteErrorCode, ServiceSqliteErrorKind, }; diff --git a/crates/service_sqlite/src/open.rs b/crates/service_sqlite/src/open.rs @@ -6,10 +6,43 @@ use std::{ path::{Path, PathBuf}, }; +#[cfg(any(target_os = "linux", target_os = "macos"))] +use std::{ + fs::File, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, +}; + +#[cfg(any(target_os = "linux", target_os = "macos"))] +use fs2::FileExt; use radroots_runtime_paths::{ InstanceId, RuntimeContext, ServiceId, default_service_instance_artifacts, }; use serde::Serialize; +#[cfg(any(target_os = "linux", target_os = "macos"))] +use sqlx::{ + ConnectOptions, Connection, Sqlite, SqliteConnection, SqlitePool, + pool::PoolConnection, + sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}, +}; + +#[cfg(any(target_os = "linux", target_os = "macos"))] +use crate::{ + ServiceSqliteConnectionOptions, ServiceSqliteError, ServiceSqliteErrorKind, WriterAuthority, +}; + +#[cfg(any(target_os = "linux", target_os = "macos"))] +const STATEMENT_CACHE_CAPACITY: usize = 100; +#[cfg(any(target_os = "linux", target_os = "macos"))] +const COMMAND_BUFFER_CAPACITY: usize = 50; +#[cfg(any(target_os = "linux", target_os = "macos"))] +const ROW_BUFFER_CAPACITY: usize = 50; +#[cfg(any(target_os = "linux", target_os = "macos"))] +const WAL_FILE_NAME: &str = "state.sqlite-wal"; +#[cfg(any(target_os = "linux", target_os = "macos"))] +const SHARED_MEMORY_FILE_NAME: &str = "state.sqlite-shm"; /// Canonical database and writer-lock paths for one validated service instance. /// @@ -146,10 +179,751 @@ impl OpenMode { } } +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps the pool private until the Step 061 host boundary" +)] +struct PrivateConnectionPool { + pool: SqlitePool, + binding: DirectoryBinding, + paths: ServiceSqlitePaths, + authority: Option<WriterAuthority>, + inspection_guard: Option<ReadOnlyInspectionGuard>, + authority_failure: Arc<AtomicBool>, + pragma_failure: Arc<AtomicBool>, +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps pool lifecycle private until the Step 061 host boundary" +)] +impl PrivateConnectionPool { + async fn acquire(&self) -> Result<PoolConnection<Sqlite>, ServiceSqliteError> { + self.binding.validate(&self.paths)?; + let result = self.pool.acquire().await; + self.binding.validate(&self.paths)?; + result.map_err(|source| { + let kind = if self.authority_failure.load(Ordering::Acquire) { + ServiceSqliteErrorKind::Authority + } else if self.pragma_failure.load(Ordering::Acquire) { + ServiceSqliteErrorKind::Pragma + } else { + ServiceSqliteErrorKind::Open + }; + connection_source(kind, source) + }) + } + + async fn close(mut self) -> Option<WriterAuthority> { + self.pool.close().await; + self.inspection_guard.take(); + self.authority.take() + } +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps pool opening private until the Step 061 host boundary" +)] +async fn open_existing_connection_pool( + paths: &ServiceSqlitePaths, + mode: OpenMode, + policy: ServiceSqliteConnectionOptions, +) -> Result<PrivateConnectionPool, ServiceSqliteError> { + if mode == OpenMode::Initialize { + return Err(connection_error( + ServiceSqliteErrorKind::Open, + ConnectionFailureKind::UnsupportedMode, + )); + } + let (authority, inspection_guard) = match mode { + OpenMode::ReadWriteExisting => (WriterAuthority::acquire(paths, mode)?, None), + OpenMode::ReadOnlyInspection => (None, Some(ReadOnlyInspectionGuard::acquire(paths)?)), + OpenMode::Initialize => unreachable!("initialize mode returned above"), + }; + open_connection_pool(paths, mode, policy, authority, inspection_guard).await +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps pool opening private until the Step 061 host boundary" +)] +async fn open_initialized_connection_pool( + paths: &ServiceSqlitePaths, + policy: ServiceSqliteConnectionOptions, + authority: WriterAuthority, +) -> Result<PrivateConnectionPool, ServiceSqliteError> { + authority.validate_for(paths)?; + open_connection_pool(paths, OpenMode::Initialize, policy, Some(authority), None).await +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps pool construction private until the Step 061 host boundary" +)] +async fn open_connection_pool( + paths: &ServiceSqlitePaths, + mode: OpenMode, + policy: ServiceSqliteConnectionOptions, + authority: Option<WriterAuthority>, + inspection_guard: Option<ReadOnlyInspectionGuard>, +) -> Result<PrivateConnectionPool, ServiceSqliteError> { + let binding = match (mode, authority.as_ref(), inspection_guard.as_ref()) { + (OpenMode::Initialize | OpenMode::ReadWriteExisting, Some(authority), None) => { + authority.validate_for(paths)?; + DirectoryBinding::capture(authority.directory(), paths)? + } + (OpenMode::ReadOnlyInspection, None, Some(inspection_guard)) => { + DirectoryBinding::capture(&inspection_guard.directory, paths)? + } + _ => { + return Err(connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + )); + } + }; + + let connect_options = sqlite_connect_options(paths, mode, policy); + binding.validate(paths)?; + let preflight_result = SqliteConnection::connect_with(&connect_options).await; + binding.validate(paths)?; + let mut preflight = preflight_result + .map_err(|source| connection_source(ServiceSqliteErrorKind::Open, source))?; + let preflight_policy = verify_connection_policy(&mut preflight, mode, policy).await; + binding.validate(paths)?; + preflight_policy.map_err(|source| connection_source(ServiceSqliteErrorKind::Pragma, source))?; + let preflight_close = preflight.close().await; + binding.validate(paths)?; + preflight_close.map_err(|source| connection_source(ServiceSqliteErrorKind::Open, source))?; + + let after_policy = policy; + let before_policy = policy; + let after_mode = mode; + let before_mode = mode; + let retained_binding = binding.clone(); + let after_binding = binding.clone(); + let before_binding = binding.clone(); + let pool_binding = binding; + let after_paths = paths.clone(); + let before_paths = paths.clone(); + let authority_failure = Arc::new(AtomicBool::new(false)); + let pragma_failure = Arc::new(AtomicBool::new(false)); + let after_authority_failure = Arc::clone(&authority_failure); + let after_pragma_failure = Arc::clone(&pragma_failure); + let before_authority_failure = Arc::clone(&authority_failure); + let before_pragma_failure = Arc::clone(&pragma_failure); + let pool_result = SqlitePoolOptions::new() + .min_connections(1) + .max_connections(policy.max_connections()) + .acquire_timeout(policy.busy_timeout()) + .idle_timeout(None) + .max_lifetime(None) + .test_before_acquire(true) + .after_connect(move |connection, _metadata| { + let binding = after_binding.clone(); + let paths = after_paths.clone(); + let authority_failure = Arc::clone(&after_authority_failure); + let pragma_failure = Arc::clone(&after_pragma_failure); + Box::pin(async move { + if binding.validate(&paths).is_err() { + authority_failure.store(true, Ordering::Release); + return Err(sqlx::Error::Protocol( + "SQLite connection authority mismatch".to_owned(), + )); + } + let policy_result = + connection_policy_matches(connection, after_mode, after_policy).await; + if binding.validate(&paths).is_err() { + authority_failure.store(true, Ordering::Release); + return Err(sqlx::Error::Protocol( + "SQLite connection authority mismatch".to_owned(), + )); + } + let matches = policy_result.inspect_err(|_| { + pragma_failure.store(true, Ordering::Release); + })?; + if !matches { + pragma_failure.store(true, Ordering::Release); + return Err(sqlx::Error::Protocol( + "SQLite connection policy mismatch".to_owned(), + )); + } + Ok(()) + }) + }) + .before_acquire(move |connection, _metadata| { + let binding = before_binding.clone(); + let paths = before_paths.clone(); + let authority_failure = Arc::clone(&before_authority_failure); + let pragma_failure = Arc::clone(&before_pragma_failure); + Box::pin(async move { + if binding.validate(&paths).is_err() { + authority_failure.store(true, Ordering::Release); + return Err(sqlx::Error::Protocol( + "SQLite connection authority mismatch".to_owned(), + )); + } + let policy_result = + connection_policy_matches(connection, before_mode, before_policy).await; + if binding.validate(&paths).is_err() { + authority_failure.store(true, Ordering::Release); + return Err(sqlx::Error::Protocol( + "SQLite connection authority mismatch".to_owned(), + )); + } + let matches = policy_result.inspect_err(|_| { + pragma_failure.store(true, Ordering::Release); + })?; + if !matches { + pragma_failure.store(true, Ordering::Release); + return Ok(false); + } + Ok(true) + }) + }) + .connect_with(connect_options) + .await; + pool_binding.validate(paths)?; + let pool = pool_result.map_err(|source| { + let kind = if authority_failure.load(Ordering::Acquire) { + ServiceSqliteErrorKind::Authority + } else if pragma_failure.load(Ordering::Acquire) { + ServiceSqliteErrorKind::Pragma + } else { + ServiceSqliteErrorKind::Open + }; + connection_source(kind, source) + })?; + + Ok(PrivateConnectionPool { + pool, + binding: retained_binding, + paths: paths.clone(), + authority, + inspection_guard, + authority_failure, + pragma_failure, + }) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps SQLx options private until the Step 061 host boundary" +)] +fn sqlite_connect_options( + paths: &ServiceSqlitePaths, + mode: OpenMode, + policy: ServiceSqliteConnectionOptions, +) -> SqliteConnectOptions { + let mut options = SqliteConnectOptions::new() + .filename(paths.state_database()) + .read_only(mode == OpenMode::ReadOnlyInspection) + .create_if_missing(false) + .foreign_keys(true) + .busy_timeout(policy.busy_timeout()) + .synchronous(SqliteSynchronous::Full) + .pragma("trusted_schema", "OFF") + .pragma( + "query_only", + if mode == OpenMode::ReadOnlyInspection { + "ON" + } else { + "OFF" + }, + ) + .statement_cache_capacity(STATEMENT_CACHE_CAPACITY) + .command_buffer_size(COMMAND_BUFFER_CAPACITY) + .row_buffer_size(ROW_BUFFER_CAPACITY) + .disable_statement_logging(); + if mode != OpenMode::ReadOnlyInspection { + options = options.journal_mode(SqliteJournalMode::Wal); + } else { + options = options.immutable(true); + } + options +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps pragma verification private until the Step 061 host boundary" +)] +async fn verify_connection_policy( + connection: &mut SqliteConnection, + mode: OpenMode, + policy: ServiceSqliteConnectionOptions, +) -> Result<(), sqlx::Error> { + if connection_policy_matches(connection, mode, policy).await? { + Ok(()) + } else { + Err(sqlx::Error::Protocol( + "SQLite connection policy mismatch".to_owned(), + )) + } +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps pragma verification private until the Step 061 host boundary" +)] +async fn connection_policy_matches( + connection: &mut SqliteConnection, + mode: OpenMode, + policy: ServiceSqliteConnectionOptions, +) -> Result<bool, sqlx::Error> { + let journal_mode = sqlx::query_scalar::<_, String>("PRAGMA journal_mode") + .fetch_one(&mut *connection) + .await?; + let synchronous = sqlx::query_scalar::<_, i64>("PRAGMA synchronous") + .fetch_one(&mut *connection) + .await?; + let foreign_keys = sqlx::query_scalar::<_, i64>("PRAGMA foreign_keys") + .fetch_one(&mut *connection) + .await?; + let trusted_schema = sqlx::query_scalar::<_, i64>("PRAGMA trusted_schema") + .fetch_one(&mut *connection) + .await?; + let busy_timeout = sqlx::query_scalar::<_, i64>("PRAGMA busy_timeout") + .fetch_one(&mut *connection) + .await?; + let query_only = sqlx::query_scalar::<_, i64>("PRAGMA query_only") + .fetch_one(&mut *connection) + .await?; + // SQLite reports `delete` for immutable handles; the inspection guard + // independently verifies WAL read/write header bytes before this opens. + let journal_mode_matches = if mode == OpenMode::ReadOnlyInspection { + journal_mode.eq_ignore_ascii_case("delete") + } else { + journal_mode.eq_ignore_ascii_case("wal") + }; + Ok(journal_mode_matches + && synchronous == 2 + && foreign_keys == 1 + && trusted_schema == 0 + && busy_timeout == policy.busy_timeout_milliseconds() + && query_only == i64::from(mode == OpenMode::ReadOnlyInspection)) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps connection failures private until the Step 061 host boundary" +)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ConnectionFailureKind { + UnsupportedMode, + AuthorityMismatch, + InspectionUnavailable, + InspectionContended, +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +impl fmt::Display for ConnectionFailureKind { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::UnsupportedMode => "SQLite initialize mode requires reserved state", + Self::AuthorityMismatch => "SQLite writer authority is missing or mismatched", + Self::InspectionUnavailable => "SQLite inspection authority is unavailable", + Self::InspectionContended => "SQLite inspection requires an offline writer", + }) + } +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +impl Error for ConnectionFailureKind {} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps connection failures private until the Step 061 host boundary" +)] +fn connection_error( + kind: ServiceSqliteErrorKind, + cause: ConnectionFailureKind, +) -> ServiceSqliteError { + ServiceSqliteError::with_source(kind, cause) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps dependency causes private until the Step 061 host boundary" +)] +fn connection_source(kind: ServiceSqliteErrorKind, cause: sqlx::Error) -> ServiceSqliteError { + ServiceSqliteError::with_source(kind, cause) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps inspection authority private until the Step 061 host boundary" +)] +struct ReadOnlyInspectionGuard { + lock: File, + directory: File, + _database: File, +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[allow( + dead_code, + reason = "Step 056 keeps inspection authority private until the Step 061 host boundary" +)] +impl ReadOnlyInspectionGuard { + fn acquire(paths: &ServiceSqlitePaths) -> Result<Self, ServiceSqliteError> { + use rustix::{ + fs::{AtFlags, FileType, Mode, OFlags, fstat, open, openat, statat}, + process::geteuid, + }; + + let directory = open( + paths + .state_lock() + .parent() + .ok_or_else(|| inspection_error(ConnectionFailureKind::InspectionUnavailable))?, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(|_| inspection_error(ConnectionFailureKind::InspectionUnavailable))?; + let directory_status = fstat(&directory) + .map_err(|_| inspection_error(ConnectionFailureKind::InspectionUnavailable))?; + if !FileType::from_raw_mode(directory_status.st_mode).is_dir() + || directory_status.st_uid != geteuid().as_raw() + || u32::from(directory_status.st_mode) & 0o022 != 0 + { + return Err(inspection_error( + ConnectionFailureKind::InspectionUnavailable, + )); + } + let lock = openat( + &directory, + radroots_runtime_paths::SERVICE_STATE_LOCK_FILE_NAME, + OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC | OFlags::NONBLOCK, + Mode::empty(), + ) + .map_err(|_| inspection_error(ConnectionFailureKind::InspectionUnavailable))?; + let lock_status = fstat(&lock) + .map_err(|_| inspection_error(ConnectionFailureKind::InspectionUnavailable))?; + if !FileType::from_raw_mode(lock_status.st_mode).is_file() + || u64::from(lock_status.st_nlink) != 1 + || lock_status.st_uid != geteuid().as_raw() + || u32::from(lock_status.st_mode) & 0o777 != 0o600 + { + return Err(inspection_error( + ConnectionFailureKind::InspectionUnavailable, + )); + } + let lock = File::from(lock); + FileExt::try_lock_shared(&lock).map_err(|error| { + if error.kind() == std::io::ErrorKind::WouldBlock { + inspection_error(ConnectionFailureKind::InspectionContended) + } else { + inspection_error(ConnectionFailureKind::InspectionUnavailable) + } + })?; + for sidecar in [WAL_FILE_NAME, SHARED_MEMORY_FILE_NAME] { + match statat(&directory, sidecar, AtFlags::SYMLINK_NOFOLLOW) { + Err(error) if error == rustix::io::Errno::NOENT => {} + Ok(_) | Err(_) => { + return Err(inspection_error( + ConnectionFailureKind::InspectionUnavailable, + )); + } + } + } + let database = openat( + &directory, + radroots_runtime_paths::SERVICE_STATE_DATABASE_FILE_NAME, + OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Open, + ConnectionFailureKind::InspectionUnavailable, + ) + })?; + let database_status = fstat(&database).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Open, + ConnectionFailureKind::InspectionUnavailable, + ) + })?; + if !FileType::from_raw_mode(database_status.st_mode).is_file() + || u64::from(database_status.st_nlink) != 1 + || database_status.st_uid != geteuid().as_raw() + || u32::from(database_status.st_mode) & 0o777 != 0o600 + { + return Err(connection_error( + ServiceSqliteErrorKind::Open, + ConnectionFailureKind::InspectionUnavailable, + )); + } + let database = File::from(database); + let mut sqlite_header = [0_u8; 20]; + std::os::unix::fs::FileExt::read_exact_at(&database, &mut sqlite_header, 0).map_err( + |_| { + connection_error( + ServiceSqliteErrorKind::Open, + ConnectionFailureKind::InspectionUnavailable, + ) + }, + )?; + if &sqlite_header[..16] != b"SQLite format 3\0" + || sqlite_header[18] != 2 + || sqlite_header[19] != 2 + { + return Err(connection_error( + ServiceSqliteErrorKind::Pragma, + ConnectionFailureKind::InspectionUnavailable, + )); + } + Ok(Self { + lock, + directory: File::from(directory), + _database: database, + }) + } +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +impl Drop for ReadOnlyInspectionGuard { + fn drop(&mut self) { + let _ = FileExt::unlock(&self.lock); + } +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn inspection_error(cause: ConnectionFailureKind) -> ServiceSqliteError { + connection_error(ServiceSqliteErrorKind::Authority, cause) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[derive(Clone)] +struct DirectoryBinding { + database_path: PathBuf, + directory: Arc<File>, + directory_device: u64, + directory_inode: u64, + database: Arc<File>, + database_device: u64, + database_inode: u64, +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +impl DirectoryBinding { + fn capture(directory: &File, paths: &ServiceSqlitePaths) -> Result<Self, ServiceSqliteError> { + use rustix::{ + fs::{FileType, Mode, OFlags, fstat, openat}, + process::geteuid, + }; + + let directory_status = fstat(directory).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let database = openat( + directory, + radroots_runtime_paths::SERVICE_STATE_DATABASE_FILE_NAME, + OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(|error| { + connection_error( + if error == rustix::io::Errno::NOENT { + ServiceSqliteErrorKind::Open + } else { + ServiceSqliteErrorKind::Authority + }, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let database_status = fstat(&database).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + if !FileType::from_raw_mode(database_status.st_mode).is_file() + || u64::from(database_status.st_nlink) != 1 + || database_status.st_uid != geteuid().as_raw() + || u32::from(database_status.st_mode) & 0o777 != 0o600 + { + return Err(connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + )); + } + Ok(Self { + database_path: paths.state_database().to_path_buf(), + directory: Arc::new(directory.try_clone().map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?), + directory_device: u64::try_from(directory_status.st_dev).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?, + directory_inode: directory_status.st_ino, + database: Arc::new(File::from(database)), + database_device: u64::try_from(database_status.st_dev).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?, + database_inode: database_status.st_ino, + }) + } + + fn validate(&self, paths: &ServiceSqlitePaths) -> Result<(), ServiceSqliteError> { + use rustix::{ + fs::{FileType, Mode, OFlags, fstat, open, openat}, + process::geteuid, + }; + + if self.database_path != paths.state_database() { + return Err(connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + )); + } + let directory = open( + paths.state_database().parent().ok_or_else(|| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let held_directory_status = fstat(&*self.directory).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let directory_status = fstat(&directory).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let directory_device = u64::try_from(directory_status.st_dev).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let held_directory_device = u64::try_from(held_directory_status.st_dev).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + if directory_device != self.directory_device + || directory_status.st_ino != self.directory_inode + || held_directory_device != self.directory_device + || held_directory_status.st_ino != self.directory_inode + { + return Err(connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + )); + } + + let database = openat( + &directory, + radroots_runtime_paths::SERVICE_STATE_DATABASE_FILE_NAME, + OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let held_database_status = fstat(&*self.database).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let database_status = fstat(&database).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let database_device = u64::try_from(database_status.st_dev).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + let held_database_device = u64::try_from(held_database_status.st_dev).map_err(|_| { + connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + ) + })?; + if !FileType::from_raw_mode(database_status.st_mode).is_file() + || u64::from(database_status.st_nlink) != 1 + || database_status.st_uid != geteuid().as_raw() + || u32::from(database_status.st_mode) & 0o777 != 0o600 + || !FileType::from_raw_mode(held_database_status.st_mode).is_file() + || u64::from(held_database_status.st_nlink) != 1 + || held_database_status.st_uid != geteuid().as_raw() + || u32::from(held_database_status.st_mode) & 0o777 != 0o600 + || database_device != self.database_device + || database_status.st_ino != self.database_inode + || held_database_device != self.database_device + || held_database_status.st_ino != self.database_inode + { + return Err(connection_error( + ServiceSqliteErrorKind::Authority, + ConnectionFailureKind::AuthorityMismatch, + )); + } + Ok(()) + } +} + #[cfg(test)] mod tests { use std::path::PathBuf; + #[cfg(any(target_os = "linux", target_os = "macos"))] + use std::{ + collections::BTreeMap, + convert::Infallible, + fs, + os::unix::fs::{PermissionsExt, symlink}, + time::{Duration, SystemTime}, + }; + use radroots_runtime_paths::{ RadrootsHostEnvironment, RadrootsPathProfile, RadrootsPathResolver, RadrootsPlatform, RuntimeContextBootstrap, RuntimeContextSource, @@ -157,6 +931,76 @@ mod tests { use super::*; + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[derive(Debug, PartialEq, Eq)] + struct FileSnapshot { + bytes: Vec<u8>, + length: u64, + modified: SystemTime, + mode: u32, + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + fn directory_snapshot(directory: &Path) -> BTreeMap<String, FileSnapshot> { + fs::read_dir(directory) + .expect("read state directory") + .map(|entry| { + let entry = entry.expect("state entry"); + let name = entry.file_name().into_string().expect("UTF-8 state entry"); + let metadata = entry.metadata().expect("state metadata"); + let snapshot = FileSnapshot { + bytes: fs::read(entry.path()).expect("state bytes"), + length: metadata.len(), + modified: metadata.modified().expect("modified time"), + mode: metadata.permissions().mode() & 0o777, + }; + (name, snapshot) + }) + .collect() + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + async fn initialized_authority( + root: &Path, + instance: &str, + ) -> (ServiceSqlitePaths, WriterAuthority) { + let paths = ServiceSqlitePaths::from_runtime_context(&runtime_context( + RadrootsPathProfile::RepoLocal, + Some(root.to_path_buf()), + "myc", + instance, + )) + .expect("SQLite paths"); + fs::create_dir_all(paths.state_database().parent().expect("state directory")) + .expect("create state directory"); + let authority = + crate::initialize_database(&paths, OpenMode::Initialize, |database_path| async move { + let options = SqliteConnectOptions::new() + .filename(database_path) + .create_if_missing(false); + let connection = SqliteConnection::connect_with(&options) + .await + .expect("open reserved database"); + connection.close().await.expect("close reserved database"); + Ok::<_, Infallible>(()) + }) + .await + .expect("initialize database"); + (paths, authority) + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + async fn initialized_pool( + root: &Path, + policy: ServiceSqliteConnectionOptions, + ) -> (ServiceSqlitePaths, PrivateConnectionPool) { + let (paths, authority) = initialized_authority(root, "primary").await; + let pool = open_initialized_connection_pool(&paths, policy, authority) + .await + .expect("open initialized pool"); + (paths, pool) + } + fn runtime_context( profile: RadrootsPathProfile, repo_local_root: Option<PathBuf>, @@ -295,4 +1139,361 @@ mod tests { assert_eq!(mode.requires_writer_authority(), requires_writer); } } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn every_connection_uses_the_exact_reviewed_pragma_policy() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::reviewed(); + let (_paths, pool) = initialized_pool(directory.path(), policy).await; + + let mut connections = Vec::with_capacity(8); + for _ in 0..8 { + connections.push(pool.acquire().await.expect("pooled connection")); + } + assert_eq!(pool.pool.size(), 8); + for connection in &mut connections { + assert!( + connection_policy_matches(connection, OpenMode::Initialize, policy) + .await + .unwrap() + ); + } + drop(connections); + assert!(pool.close().await.expect("writer authority").is_held()); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn reused_connection_drift_is_rejected_before_checkout() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::reviewed(); + let (_paths, pool) = initialized_pool(directory.path(), policy).await; + + let mut connection = pool.acquire().await.expect("pooled connection"); + sqlx::query("PRAGMA foreign_keys = OFF") + .execute(&mut *connection) + .await + .expect("drift pragma"); + assert_eq!( + sqlx::query_scalar::<_, i64>("PRAGMA foreign_keys") + .fetch_one(&mut *connection) + .await + .unwrap(), + 0 + ); + drop(connection); + + let mut replacement = pool.acquire().await.expect("replacement connection"); + assert!( + connection_policy_matches(&mut replacement, OpenMode::Initialize, policy) + .await + .unwrap() + ); + drop(replacement); + let _authority = pool.close().await; + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn existing_modes_never_create_missing_state_and_enforce_authority() { + let directory = tempfile::tempdir().expect("temporary directory"); + let paths = ServiceSqlitePaths::from_runtime_context(&runtime_context( + RadrootsPathProfile::RepoLocal, + Some(directory.path().to_path_buf()), + "rhi", + "default", + )) + .expect("SQLite paths"); + fs::create_dir_all(paths.state_database().parent().expect("state directory")) + .expect("create state directory"); + + for mode in [OpenMode::ReadWriteExisting, OpenMode::ReadOnlyInspection] { + let result = open_existing_connection_pool( + &paths, + mode, + ServiceSqliteConnectionOptions::reviewed(), + ) + .await; + let Err(error) = result else { + panic!("missing database must fail"); + }; + assert_eq!(error.kind(), ServiceSqliteErrorKind::Open); + assert!(!paths.state_database().exists()); + } + let result = open_existing_connection_pool( + &paths, + OpenMode::Initialize, + ServiceSqliteConnectionOptions::reviewed(), + ) + .await; + let Err(error) = result else { + panic!("initialize needs reserved state"); + }; + assert_eq!(error.kind(), ServiceSqliteErrorKind::Open); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn initialized_pool_rejects_mismatched_paths_and_rebound_directory() { + let directory = tempfile::tempdir().expect("temporary directory"); + let (_paths, authority) = initialized_authority(directory.path(), "primary").await; + let other = ServiceSqlitePaths::from_runtime_context(&runtime_context( + RadrootsPathProfile::RepoLocal, + Some(directory.path().to_path_buf()), + "myc", + "secondary", + )) + .expect("other SQLite paths"); + fs::create_dir_all( + other + .state_database() + .parent() + .expect("other state directory"), + ) + .expect("create other state directory"); + let result = open_initialized_connection_pool( + &other, + ServiceSqliteConnectionOptions::reviewed(), + authority, + ) + .await; + let Err(error) = result else { + panic!("mismatched paths must fail"); + }; + assert_eq!(error.kind(), ServiceSqliteErrorKind::Authority); + + let (paths, authority) = initialized_authority(directory.path(), "rebound").await; + let state_directory = paths.state_database().parent().expect("state directory"); + let displaced = directory.path().join("displaced-state"); + fs::rename(state_directory, &displaced).expect("displace state directory"); + fs::create_dir_all(state_directory).expect("replace state directory"); + fs::copy(displaced.join("state.sqlite"), paths.state_database()) + .expect("copy replacement database"); + let result = open_initialized_connection_pool( + &paths, + ServiceSqliteConnectionOptions::reviewed(), + authority, + ) + .await; + let Err(error) = result else { + panic!("rebound state directory must fail"); + }; + assert_eq!(error.kind(), ServiceSqliteErrorKind::Authority); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn pool_checkout_rejects_state_directory_replacement_before_growth() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::new(Duration::from_millis(500), 2).unwrap(); + let (paths, pool) = initialized_pool(directory.path(), policy).await; + assert_eq!(pool.pool.size(), 1); + + let state_directory = paths.state_database().parent().expect("state directory"); + let displaced = directory.path().join("displaced-live-state"); + fs::rename(state_directory, &displaced).expect("displace live state directory"); + fs::create_dir_all(state_directory).expect("replace live state directory"); + fs::copy(displaced.join("state.sqlite"), paths.state_database()) + .expect("copy replacement database"); + + let error = pool + .acquire() + .await + .expect_err("rebound directory must prevent checkout and pool growth"); + assert_eq!(error.kind(), ServiceSqliteErrorKind::Authority); + assert_eq!(pool.pool.size(), 1); + let authority = pool.close().await.expect("writer authority retained"); + assert!(authority.is_held()); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn writable_existing_rejects_database_symlink_and_hardlink() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::reviewed(); + + let (symlink_paths, symlink_authority) = + initialized_authority(directory.path(), "symlink-database").await; + drop(symlink_authority); + let symlink_backing = symlink_paths + .state_database() + .parent() + .expect("state directory") + .join("backing.sqlite"); + fs::rename(symlink_paths.state_database(), &symlink_backing) + .expect("displace symlink database"); + symlink(&symlink_backing, symlink_paths.state_database()).expect("database symlink"); + let symlink_result = + open_existing_connection_pool(&symlink_paths, OpenMode::ReadWriteExisting, policy) + .await; + let Err(symlink_error) = symlink_result else { + panic!("database symlink must fail"); + }; + assert_eq!(symlink_error.kind(), ServiceSqliteErrorKind::Authority); + + let (hardlink_paths, hardlink_authority) = + initialized_authority(directory.path(), "hardlink-database").await; + drop(hardlink_authority); + let hardlink_alias = hardlink_paths + .state_database() + .parent() + .expect("state directory") + .join("alias.sqlite"); + fs::hard_link(hardlink_paths.state_database(), hardlink_alias).expect("database hard link"); + let hardlink_result = + open_existing_connection_pool(&hardlink_paths, OpenMode::ReadWriteExisting, policy) + .await; + let Err(hardlink_error) = hardlink_result else { + panic!("database hard link must fail"); + }; + assert_eq!(hardlink_error.kind(), ServiceSqliteErrorKind::Authority); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn lazy_pool_growth_rejects_same_directory_database_replacement() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::new(Duration::from_millis(500), 2).unwrap(); + let (paths, pool) = initialized_pool(directory.path(), policy).await; + let held = pool.acquire().await.expect("hold initial connection"); + + let displaced = paths + .state_database() + .parent() + .expect("state directory") + .join("displaced.sqlite"); + fs::rename(paths.state_database(), &displaced).expect("displace live database"); + fs::copy(&displaced, paths.state_database()).expect("copy replacement database"); + fs::set_permissions(paths.state_database(), fs::Permissions::from_mode(0o600)) + .expect("secure replacement database"); + + let error = pool + .acquire() + .await + .expect_err("database replacement must prevent lazy pool growth"); + assert_eq!(error.kind(), ServiceSqliteErrorKind::Authority); + drop(held); + let authority = pool.close().await.expect("writer authority retained"); + assert!(authority.is_held()); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn read_only_inspection_is_offline_query_only_and_side_effect_free() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::reviewed(); + let (paths, writable) = initialized_pool(directory.path(), policy).await; + let mut connection = writable.acquire().await.expect("writable connection"); + sqlx::query("CREATE TABLE inspection_fixture (value INTEGER NOT NULL)") + .execute(&mut *connection) + .await + .expect("create fixture"); + sqlx::query("INSERT INTO inspection_fixture (value) VALUES (41)") + .execute(&mut *connection) + .await + .expect("insert fixture"); + drop(connection); + + let contended = + open_existing_connection_pool(&paths, OpenMode::ReadOnlyInspection, policy).await; + let Err(error) = contended else { + panic!("inspection must reject an active writer"); + }; + assert_eq!(error.kind(), ServiceSqliteErrorKind::Authority); + + let authority = writable.close().await.expect("writer authority"); + drop(authority); + let state_directory = paths.state_database().parent().expect("state directory"); + let stale_wal = state_directory.join(WAL_FILE_NAME); + fs::write(&stale_wal, b"stale-wal-evidence").expect("write stale WAL evidence"); + let stale = + open_existing_connection_pool(&paths, OpenMode::ReadOnlyInspection, policy).await; + let Err(error) = stale else { + panic!("inspection must reject stale WAL state"); + }; + assert_eq!(error.kind(), ServiceSqliteErrorKind::Authority); + assert_eq!(fs::read(&stale_wal).unwrap(), b"stale-wal-evidence"); + fs::remove_file(stale_wal).expect("remove test WAL evidence"); + let before = directory_snapshot(state_directory); + + let read_only = open_existing_connection_pool(&paths, OpenMode::ReadOnlyInspection, policy) + .await + .expect("offline read-only inspection"); + let mut connection = read_only.acquire().await.expect("inspection connection"); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT value FROM inspection_fixture") + .fetch_one(&mut *connection) + .await + .expect("read fixture"), + 41 + ); + assert!( + sqlx::query("INSERT INTO inspection_fixture (value) VALUES (42)") + .execute(&mut *connection) + .await + .is_err() + ); + assert!( + connection_policy_matches(&mut connection, OpenMode::ReadOnlyInspection, policy) + .await + .unwrap() + ); + drop(connection); + assert!(read_only.close().await.is_none()); + + let after = directory_snapshot(state_directory); + assert_eq!( + after.keys().collect::<Vec<_>>(), + before.keys().collect::<Vec<_>>() + ); + for (name, before_file) in &before { + let after_file = after.get(name).expect("same state entry"); + assert!( + after_file.bytes == before_file.bytes, + "{name} bytes changed" + ); + assert_eq!( + after_file.length, before_file.length, + "{name} length changed" + ); + assert_eq!( + after_file.modified, before_file.modified, + "{name} mtime changed" + ); + assert_eq!(after_file.mode, before_file.mode, "{name} mode changed"); + } + assert!(!after.contains_key("state.sqlite-wal")); + assert!(!after.contains_key("state.sqlite-shm")); + } + + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[tokio::test(flavor = "current_thread")] + async fn pool_saturation_recovers_and_explicit_close_finishes() { + let directory = tempfile::tempdir().expect("temporary directory"); + let policy = ServiceSqliteConnectionOptions::new(Duration::from_millis(500), 1).unwrap(); + let (paths, pool) = initialized_pool(directory.path(), policy).await; + let observer = pool.pool.clone(); + let held = pool.acquire().await.expect("only connection"); + let saturated = pool.pool.try_acquire(); + assert!(saturated.is_none()); + drop(held); + let recovered = pool.acquire().await.expect("recovered connection"); + drop(recovered); + + let authority = pool.close().await.expect("writer authority retained"); + assert!(observer.is_closed()); + assert!(authority.is_held()); + drop(authority); + + let read_only = open_existing_connection_pool( + &paths, + OpenMode::ReadOnlyInspection, + ServiceSqliteConnectionOptions::reviewed(), + ) + .await + .expect("read-only inspection"); + assert!(read_only.authority.is_none()); + assert!(read_only.close().await.is_none()); + } } diff --git a/crates/service_sqlite/tests/package_boundary.rs b/crates/service_sqlite/tests/package_boundary.rs @@ -3,6 +3,7 @@ use std::collections::BTreeSet; const MANIFEST: &str = include_str!("../Cargo.toml"); const ROOT: &str = include_str!("../src/lib.rs"); const AUTHORITY_SOURCE: &str = include_str!("../src/authority.rs"); +const CONFIG_SOURCE: &str = include_str!("../src/config.rs"); const ERROR_SOURCE: &str = include_str!("../src/error.rs"); const INITIALIZE_SOURCE: &str = include_str!("../src/initialize.rs"); const OPEN_SOURCE: &str = include_str!("../src/open.rs"); @@ -24,21 +25,32 @@ fn service_sqlite_is_unpublished_lint_governed_and_dependency_bounded() { assert_eq!( dependency_keys(MANIFEST, "[dependencies]"), - BTreeSet::from(["fs2", "radroots_runtime_paths", "rustix", "serde"]) + BTreeSet::from(["fs2", "radroots_runtime_paths", "rustix", "serde", "sqlx"]) ); assert_eq!( dependency_keys(MANIFEST, "[dev-dependencies]"), - BTreeSet::from(["serde_json", "tempfile"]) + BTreeSet::from(["serde_json", "tempfile", "tokio"]) ); assert_eq!( private_modules(ROOT), - BTreeSet::from(["authority", "error", "initialize", "open", "status"]) + BTreeSet::from([ + "authority", + "config", + "error", + "initialize", + "open", + "status" + ]) ); assert!(public_modules(ROOT).is_empty()); let authority_production = AUTHORITY_SOURCE .split_once("#[cfg(all(test") .map(|(production, _)| production) .expect("authority source must keep tests separated"); + let open_production = OPEN_SOURCE + .split_once("#[cfg(test)]") + .map(|(production, _)| production) + .expect("open source must keep tests separated"); for required in [ "ServiceSqliteErrorCode", @@ -46,6 +58,8 @@ fn service_sqlite_is_unpublished_lint_governed_and_dependency_bounded() { "SafeServiceSqliteError", "ServiceSqliteError", "WriterAuthority", + "ServiceSqliteConnectionOptions", + "ServiceSqliteConnectionOptionsError", "initialize_database", "ServiceSqlitePathError", "ServiceSqlitePaths", @@ -62,25 +76,71 @@ fn service_sqlite_is_unpublished_lint_governed_and_dependency_bounded() { for forbidden in [ "radroots_service_host", - "sqlx", "rusqlite", "tokio", "std::fs", "OpenOptions", "create_dir", - "Connection", - "Pool", ] { assert!( !authority_production.contains(forbidden) && !ERROR_SOURCE.contains(forbidden) - && !OPEN_SOURCE.contains(forbidden) + && !CONFIG_SOURCE.contains(forbidden) && !STATUS_SOURCE.contains(forbidden), "service SQLite source contains deferred surface `{forbidden}`" ); } for required in [ + "journal_mode(SqliteJournalMode::Wal)", + "synchronous(SqliteSynchronous::Full)", + ".foreign_keys(true)", + ".pragma(\"trusted_schema\", \"OFF\")", + ".create_if_missing(false)", + ".statement_cache_capacity(STATEMENT_CACHE_CAPACITY)", + ".command_buffer_size(COMMAND_BUFFER_CAPACITY)", + ".row_buffer_size(ROW_BUFFER_CAPACITY)", + ".immutable(true)", + "\"query_only\"", + "sqlite_header[18] != 2", + "sqlite_header[19] != 2", + "WAL_FILE_NAME", + "SHARED_MEMORY_FILE_NAME", + ".min_connections(1)", + ".max_connections(policy.max_connections())", + ".acquire_timeout(policy.busy_timeout())", + ".idle_timeout(None)", + ".max_lifetime(None)", + ".after_connect", + ".before_acquire", + "PRAGMA busy_timeout", + ] { + assert!( + OPEN_SOURCE.contains(required), + "Step 056 connection source is missing `{required}`" + ); + } + + for forbidden in [ + "pub use sqlx", + "pub use sqlx::SqliteConnection", + "pub use sqlx::SqlitePool", + "pub use sqlx::Pool", + "pub struct PrivateConnectionPool", + "pub fn open_connection_pool", + "pub async fn open_connection_pool", + "tokio::runtime", + "Runtime::new", + ] { + assert!( + !ROOT.contains(forbidden) + && !CONFIG_SOURCE.contains(forbidden) + && !OPEN_SOURCE.contains(forbidden), + "Step 056 source exposes forbidden pool/runtime surface `{forbidden}`" + ); + } + + for required in [ "OFlags::EXCL", "OFlags::NOFOLLOW", "OFlags::CLOEXEC", @@ -127,6 +187,8 @@ fn service_sqlite_is_unpublished_lint_governed_and_dependency_bounded() { "st_nlink", "fchmod", "pub fn release(&mut self)", + "database_path: paths.state_database().to_path_buf()", + "pub(crate) fn validate_for", ] { assert!( AUTHORITY_SOURCE.contains(required), @@ -135,7 +197,6 @@ fn service_sqlite_is_unpublished_lint_governed_and_dependency_bounded() { } for forbidden in [ - "state_database()", "remove_file", "create_dir", "set_len", @@ -162,12 +223,9 @@ fn service_sqlite_is_unpublished_lint_governed_and_dependency_bounded() { "create_dir", "File::open", "OpenOptions", - "fs2", - "rustix", - "lock_exclusive", ] { assert!( - !OPEN_SOURCE.contains(forbidden), + !open_production.contains(forbidden), "Step 053 open source contains deferred or forgeable surface `{forbidden}`" ); }