use crossbeam_queue::ArrayQueue;
use parking_lot::Mutex;
use rusqlite::{Connection, OpenFlags};
use std::fs;
use std::ops::{Deref, DerefMut};
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use std::thread;
use std::time::{Duration, Instant};
use crate::error::SqliteError;
use crate::writer_task::WriterTaskHandle;
use khive_storage::error::StorageError;
use khive_storage::tx_registry::{DbIdentity, TxOrigin};
const CACHE_SIZE_KIB: &str = "-65536";
const MMAP_SIZE_BYTES: &str = "1073741824";
const DEFAULT_READER_CAP: usize = 8;
const DEFAULT_WAL_AUTOCHECKPOINT_PAGES: u32 = 4000;
const DEFAULT_JOURNAL_SIZE_LIMIT_BYTES: i64 = 67_108_864; const DEFAULT_WRITE_QUEUE_CAPACITY: usize = 256;
#[derive(Clone, Debug)]
pub struct PoolConfig {
pub path: Option<PathBuf>,
pub max_readers: usize,
pub wal_mode: bool,
pub busy_timeout: Duration,
pub checkout_timeout: Duration,
pub wal_autocheckpoint_pages: u32,
pub journal_size_limit_bytes: i64,
pub read_only: bool,
pub write_queue_enabled: bool,
pub write_queue_capacity: usize,
}
impl Default for PoolConfig {
fn default() -> Self {
Self {
path: None,
max_readers: std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
.clamp(1, DEFAULT_READER_CAP),
wal_mode: true,
busy_timeout: Duration::from_secs(
std::env::var("KHIVE_BUSY_TIMEOUT_SECS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(30),
),
checkout_timeout: Duration::from_secs(
std::env::var("KHIVE_CHECKOUT_TIMEOUT_SECS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(5),
),
wal_autocheckpoint_pages: std::env::var("KHIVE_WAL_AUTOCHECKPOINT_PAGES")
.ok()
.and_then(|v| v.parse::<u32>().ok())
.unwrap_or(DEFAULT_WAL_AUTOCHECKPOINT_PAGES),
journal_size_limit_bytes: std::env::var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES")
.ok()
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(DEFAULT_JOURNAL_SIZE_LIMIT_BYTES),
read_only: false,
write_queue_enabled: std::env::var("KHIVE_WRITE_QUEUE")
.map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
.unwrap_or(false),
write_queue_capacity: std::env::var("KHIVE_WRITE_QUEUE_CAPACITY")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|&n| n > 0)
.unwrap_or(DEFAULT_WRITE_QUEUE_CAPACITY),
}
}
}
pub struct ConnectionPool {
writer: Arc<Mutex<Connection>>,
readers: ArrayQueue<Connection>,
max_readers: usize,
config: PoolConfig,
writer_task: OnceLock<Option<WriterTaskHandle>>,
origin: TxOrigin,
identity_path: Option<PathBuf>,
#[cfg(test)]
writer_task_spawn_count: std::sync::atomic::AtomicUsize,
}
enum ReaderLease<'pool> {
Pooled(Connection),
Shared(parking_lot::MutexGuard<'pool, Connection>),
}
pub struct ReaderGuard<'pool> {
lease: Option<ReaderLease<'pool>>,
pool: &'pool ConnectionPool,
}
impl<'pool> ReaderGuard<'pool> {
pub fn conn(&self) -> &Connection {
match self
.lease
.as_ref()
.expect("reader guard missing connection")
{
ReaderLease::Pooled(conn) => conn,
ReaderLease::Shared(guard) => guard,
}
}
}
impl<'pool> Deref for ReaderGuard<'pool> {
type Target = Connection;
fn deref(&self) -> &Self::Target {
self.conn()
}
}
impl<'pool> Drop for ReaderGuard<'pool> {
fn drop(&mut self) {
let Some(lease) = self.lease.take() else {
return;
};
match lease {
ReaderLease::Pooled(conn) => self.pool.return_reader(conn),
ReaderLease::Shared(_guard) => {}
}
}
}
pub struct WriterGuard<'pool> {
guard: parking_lot::MutexGuard<'pool, Connection>,
origin: TxOrigin,
}
impl<'pool> WriterGuard<'pool> {
pub fn conn(&self) -> &Connection {
&self.guard
}
pub fn conn_mut(&mut self) -> &mut Connection {
&mut self.guard
}
pub fn transaction<F, R>(&self, f: F) -> Result<R, SqliteError>
where
F: FnOnce(&Connection) -> Result<R, SqliteError>,
{
self.guard.execute_batch("BEGIN IMMEDIATE")?;
let _tx_handle = khive_storage::tx_registry::register_scoped(
Some("writer_guard_tx".to_string()),
self.origin.clone(),
);
match f(&self.guard) {
Ok(result) => {
if let Err(err) = self.guard.execute_batch("COMMIT") {
let _ = self.guard.execute_batch("ROLLBACK");
return Err(err.into());
}
Ok(result)
}
Err(err) => {
let _ = self.guard.execute_batch("ROLLBACK");
Err(err)
}
}
}
}
impl<'pool> Deref for WriterGuard<'pool> {
type Target = Connection;
fn deref(&self) -> &Self::Target {
self.conn()
}
}
impl<'pool> DerefMut for WriterGuard<'pool> {
fn deref_mut(&mut self) -> &mut Self::Target {
self.conn_mut()
}
}
impl ConnectionPool {
pub fn new(config: PoolConfig) -> Result<Self, SqliteError> {
let writer = open_writer_connection(&config)?;
let wal_enabled = configure_writer_connection(&writer, &config)?;
let max_readers = effective_reader_count(&config, wal_enabled);
let readers = ArrayQueue::new(max_readers.max(1));
let (origin, identity_path) = match config.path.as_ref() {
Some(path) => {
let (identity, canonical) = mint_db_identity(path)?;
(TxOrigin::Database(identity), Some(canonical))
}
None => (TxOrigin::Memory, None),
};
let pool = Self {
writer: Arc::new(Mutex::new(writer)),
readers,
max_readers,
config,
writer_task: OnceLock::new(),
origin,
identity_path,
#[cfg(test)]
writer_task_spawn_count: std::sync::atomic::AtomicUsize::new(0),
};
for _ in 0..pool.max_readers {
let conn = pool.open_reader_connection()?;
pool.readers
.push(conn)
.expect("reader queue must have capacity during pool initialization");
}
Ok(pool)
}
pub fn reader(&self) -> Result<ReaderGuard<'_>, SqliteError> {
if self.max_readers == 0 {
return Ok(ReaderGuard {
lease: Some(ReaderLease::Shared(self.writer.lock())),
pool: self,
});
}
let started = Instant::now();
let mut attempt = 0u32;
loop {
if let Some(conn) = self.readers.pop() {
return Ok(ReaderGuard {
lease: Some(ReaderLease::Pooled(conn)),
pool: self,
});
}
if started.elapsed() >= self.config.checkout_timeout {
return Err(pool_exhausted_error(
self.config.checkout_timeout,
self.max_readers,
));
}
match attempt {
0..=7 => {
let spins = 1usize << attempt;
for _ in 0..spins {
std::hint::spin_loop();
}
}
8..=15 => thread::yield_now(),
_ => {
let remaining = self
.config
.checkout_timeout
.saturating_sub(started.elapsed());
let sleep = Duration::from_micros(50 * (1u64 << (attempt - 16).min(6)));
thread::sleep(sleep.min(remaining).min(Duration::from_millis(2)));
}
}
attempt = attempt.saturating_add(1);
}
}
pub fn writer(&self) -> Result<WriterGuard<'_>, SqliteError> {
let guard = self
.writer
.try_lock_for(self.config.checkout_timeout)
.ok_or_else(|| {
SqliteError::InvalidData(format!(
"timed out after {:?} waiting for sqlite writer connection",
self.config.checkout_timeout
))
})?;
Ok(WriterGuard {
guard,
origin: self.origin(),
})
}
pub fn try_writer(&self) -> Result<WriterGuard<'_>, SqliteError> {
self.writer()
}
pub fn try_writer_nowait(&self) -> Result<WriterGuard<'_>, SqliteError> {
let guard = self.writer.try_lock().ok_or_else(|| {
SqliteError::InvalidData(
"writer connection busy (checkpoint skipped this tick)".to_string(),
)
})?;
Ok(WriterGuard {
guard,
origin: self.origin(),
})
}
pub fn available_readers(&self) -> usize {
self.readers.len()
}
pub fn max_readers(&self) -> usize {
self.max_readers
}
pub fn config(&self) -> &PoolConfig {
&self.config
}
pub fn origin(&self) -> TxOrigin {
self.origin.clone()
}
pub fn canonical_path(&self) -> Option<&Path> {
self.identity_path.as_deref()
}
pub fn writer_task_handle(&self) -> Result<Option<WriterTaskHandle>, StorageError> {
if !self.config.write_queue_enabled {
return Ok(None);
}
if let Some(existing) = self.writer_task.get() {
return Ok(existing.clone());
}
if tokio::runtime::Handle::try_current().is_err() {
return Err(StorageError::WriterTaskNoRuntime);
}
Ok(self
.writer_task
.get_or_init(|| {
#[cfg(test)]
self.writer_task_spawn_count
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
match crate::writer_task::spawn(self, self.config.write_queue_capacity) {
Ok(handle) => Some(handle),
Err(e) => {
tracing::warn!(
error = %e,
"KHIVE_WRITE_QUEUE=1 but the writer task failed to spawn; \
writes fall back to the pool-mutex path"
);
None
}
}
})
.clone())
}
#[cfg(test)]
pub(crate) fn writer_task_spawn_count(&self) -> usize {
self.writer_task_spawn_count
.load(std::sync::atomic::Ordering::SeqCst)
}
pub fn legacy_conn(&self) -> Arc<Mutex<Connection>> {
Arc::clone(&self.writer)
}
fn open_reader_connection(&self) -> Result<Connection, SqliteError> {
let path = self
.config
.path
.as_ref()
.expect("reader connections require a file-backed database");
open_reader_connection(path, &self.config)
}
pub fn open_standalone_writer(&self) -> Result<Connection, SqliteError> {
let path = self.config.path.as_ref().ok_or_else(|| {
SqliteError::InvalidData(
"in-memory databases do not support standalone connections".to_string(),
)
})?;
if self.config.read_only {
return Err(SqliteError::InvalidData(
"database is read-only: standalone write connections are not permitted".to_string(),
));
}
let conn = Connection::open_with_flags(
path,
OpenFlags::SQLITE_OPEN_READ_WRITE
| OpenFlags::SQLITE_OPEN_NO_MUTEX
| OpenFlags::SQLITE_OPEN_URI,
)?;
conn.busy_timeout(self.config.busy_timeout)?;
conn.pragma_update(None, "foreign_keys", "ON")?;
conn.pragma_update(None, "synchronous", "NORMAL")?;
Ok(conn)
}
pub fn open_standalone_reader(&self) -> Result<Connection, SqliteError> {
let path = self.config.path.as_ref().ok_or_else(|| {
SqliteError::InvalidData(
"in-memory databases do not support standalone connections".to_string(),
)
})?;
let conn = Connection::open_with_flags(
path,
OpenFlags::SQLITE_OPEN_READ_ONLY
| OpenFlags::SQLITE_OPEN_NO_MUTEX
| OpenFlags::SQLITE_OPEN_URI,
)?;
conn.busy_timeout(self.config.busy_timeout)?;
conn.pragma_update(None, "foreign_keys", "ON")?;
conn.pragma_update(None, "synchronous", "NORMAL")?;
Ok(conn)
}
fn return_reader(&self, conn: Connection) {
if self.max_readers == 0 {
return;
}
let conn = if reset_reader_connection(&conn) && reader_connection_is_healthy(&conn) {
Some(conn)
} else {
close_connection_quietly(conn);
self.open_reader_connection().ok()
};
if let Some(conn) = conn {
if let Err(conn) = self.readers.push(conn) {
eprintln!(
"[sqlite-pool] reader pool queue full, discarding replacement connection"
);
close_connection_quietly(conn);
}
}
}
}
const MAX_SYMLINK_DEPTH: u32 = 40;
fn mint_db_identity(configured_path: &Path) -> Result<(DbIdentity, PathBuf), SqliteError> {
let absolute = if configured_path.is_absolute() {
configured_path.to_path_buf()
} else {
let cwd = std::env::current_dir().map_err(|e| {
SqliteError::InvalidData(format!(
"cannot mint database identity for {configured_path:?}: failed to resolve the \
process current directory: {e}"
))
})?;
cwd.join(configured_path)
};
if absolute.exists() {
let canonical = absolute.canonicalize().map_err(|e| {
SqliteError::InvalidData(format!(
"cannot mint database identity: failed to canonicalize existing path \
{absolute:?}: {e}"
))
})?;
return Ok((
DbIdentity::new(canonical.clone().into_os_string()),
canonical,
));
}
let resolved_target = resolve_symlink_chain(&absolute)?;
let parent = resolved_target.parent().ok_or_else(|| {
SqliteError::InvalidData(format!(
"cannot mint database identity for {resolved_target:?}: path has no parent \
directory"
))
})?;
let file_name = resolved_target.file_name().ok_or_else(|| {
SqliteError::InvalidData(format!(
"cannot mint database identity for {resolved_target:?}: path has no file name"
))
})?;
let canonical_parent = parent.canonicalize().map_err(|e| {
SqliteError::InvalidData(format!(
"cannot mint database identity: parent directory {parent:?} of first-open path \
{resolved_target:?} does not exist or is inaccessible: {e}"
))
})?;
let mut identity_path = canonical_parent;
identity_path.push(file_name);
Ok((
DbIdentity::new(identity_path.clone().into_os_string()),
identity_path,
))
}
fn resolve_symlink_chain(path: &Path) -> Result<PathBuf, SqliteError> {
let mut current = path.to_path_buf();
for _ in 0..MAX_SYMLINK_DEPTH {
match fs::symlink_metadata(¤t) {
Ok(meta) if meta.file_type().is_symlink() => {
let target = fs::read_link(¤t).map_err(|e| {
SqliteError::InvalidData(format!(
"cannot mint database identity: failed to read symlink {current:?}: {e}"
))
})?;
current = if target.is_absolute() {
target
} else {
match current.parent() {
Some(parent) => parent.join(&target),
None => target,
}
};
}
_ => return Ok(current),
}
}
Err(SqliteError::InvalidData(format!(
"cannot mint database identity for {path:?}: symlink chain exceeds \
{MAX_SYMLINK_DEPTH} levels"
)))
}
fn effective_reader_count(config: &PoolConfig, wal_enabled: bool) -> usize {
if config.path.is_some() && config.wal_mode && wal_enabled {
config.max_readers
} else {
0
}
}
fn open_writer_connection(config: &PoolConfig) -> Result<Connection, SqliteError> {
match config.path.as_ref() {
Some(path) => {
let flags = if config.read_only {
writer_read_only_open_flags()
} else {
writer_open_flags()
};
Connection::open_with_flags(path, flags).map_err(Into::into)
}
None => Connection::open_in_memory().map_err(Into::into),
}
}
fn open_reader_connection(path: &Path, config: &PoolConfig) -> Result<Connection, SqliteError> {
let conn = Connection::open_with_flags(path, reader_open_flags())?;
configure_reader_connection(&conn, config)?;
Ok(conn)
}
fn writer_open_flags() -> OpenFlags {
OpenFlags::SQLITE_OPEN_READ_WRITE
| OpenFlags::SQLITE_OPEN_CREATE
| OpenFlags::SQLITE_OPEN_URI
| OpenFlags::SQLITE_OPEN_NO_MUTEX
}
fn writer_read_only_open_flags() -> OpenFlags {
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX
}
fn reader_open_flags() -> OpenFlags {
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX
}
fn configure_writer_connection(
conn: &Connection,
config: &PoolConfig,
) -> Result<bool, SqliteError> {
if config.read_only {
conn.pragma_update(None, "foreign_keys", "ON")?;
conn.busy_timeout(config.busy_timeout)?;
conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
conn.pragma_update(None, "temp_store", "MEMORY")?;
conn.pragma_update(None, "query_only", "ON")?;
let wal_enabled =
config.wal_mode && current_journal_mode(conn)?.eq_ignore_ascii_case("wal");
return Ok(wal_enabled);
}
let wants_wal = config.path.is_some() && config.wal_mode;
if wants_wal {
conn.pragma_update(None, "journal_mode", "WAL")?;
}
conn.pragma_update(None, "synchronous", "NORMAL")?;
conn.pragma_update(None, "foreign_keys", "ON")?;
conn.busy_timeout(config.busy_timeout)?;
conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
conn.pragma_update(None, "temp_store", "MEMORY")?;
let wal_enabled = wants_wal && current_journal_mode(conn)?.eq_ignore_ascii_case("wal");
if wal_enabled {
conn.pragma_update(None, "wal_autocheckpoint", config.wal_autocheckpoint_pages)?;
conn.pragma_update(None, "journal_size_limit", config.journal_size_limit_bytes)?;
}
Ok(wal_enabled)
}
fn configure_reader_connection(conn: &Connection, config: &PoolConfig) -> Result<(), SqliteError> {
conn.pragma_update(None, "foreign_keys", "ON")?;
conn.busy_timeout(config.busy_timeout)?;
conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
conn.pragma_update(None, "temp_store", "MEMORY")?;
Ok(())
}
fn current_journal_mode(conn: &Connection) -> Result<String, SqliteError> {
conn.pragma_query_value(None, "journal_mode", |row| row.get::<_, String>(0))
.map(|mode| mode.to_ascii_lowercase())
.map_err(Into::into)
}
fn reset_reader_connection(conn: &Connection) -> bool {
if conn.is_autocommit() {
return true;
}
match conn.execute_batch("ROLLBACK") {
Ok(()) => conn.is_autocommit(),
Err(rusqlite::Error::SqliteFailure(err, _)) => {
if matches!(
err.code,
rusqlite::ErrorCode::CannotOpen
| rusqlite::ErrorCode::DatabaseCorrupt
| rusqlite::ErrorCode::NotADatabase
| rusqlite::ErrorCode::DiskFull
) {
return false;
}
conn.is_autocommit()
}
Err(_) => false,
}
}
fn reader_connection_is_healthy(conn: &Connection) -> bool {
match conn.query_row("SELECT 1", [], |row| row.get::<_, i64>(0)) {
Ok(_) => true,
Err(rusqlite::Error::SqliteFailure(err, _)) => !matches!(
err.code,
rusqlite::ErrorCode::CannotOpen
| rusqlite::ErrorCode::NotADatabase
| rusqlite::ErrorCode::DatabaseCorrupt
| rusqlite::ErrorCode::PermissionDenied
| rusqlite::ErrorCode::SystemIoFailure
),
Err(_) => true,
}
}
fn close_connection_quietly(conn: Connection) {
match conn.close() {
Ok(()) => {}
Err((conn, _)) => drop(conn),
}
}
fn pool_exhausted_error(timeout: Duration, max_readers: usize) -> SqliteError {
rusqlite::Error::SqliteFailure(
rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
Some(format!(
"Pool exhausted: no reader available after {timeout:?} (max_readers={max_readers})"
)),
)
.into()
}
#[cfg(test)]
mod tests {
use super::*;
use serial_test::serial;
struct CwdGuard {
original: PathBuf,
}
impl CwdGuard {
fn enter(dir: &Path) -> Self {
let original = std::env::current_dir().unwrap();
std::env::set_current_dir(dir).unwrap();
Self { original }
}
}
impl Drop for CwdGuard {
fn drop(&mut self) {
let _ = std::env::set_current_dir(&self.original);
}
}
#[test]
#[serial]
fn pool_config_default_values_match_constants() {
let cfg = PoolConfig::default();
assert_eq!(
cfg.wal_autocheckpoint_pages,
DEFAULT_WAL_AUTOCHECKPOINT_PAGES
);
assert_eq!(
cfg.journal_size_limit_bytes,
DEFAULT_JOURNAL_SIZE_LIMIT_BYTES
);
assert_eq!(cfg.busy_timeout, Duration::from_secs(30));
assert_eq!(cfg.checkout_timeout, Duration::from_secs(5));
}
#[test]
#[serial]
fn pool_config_env_override_wal_autocheckpoint() {
std::env::set_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES", "8000");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES");
assert_eq!(cfg.wal_autocheckpoint_pages, 8000);
}
#[test]
#[serial]
fn pool_config_env_override_journal_size_limit() {
std::env::set_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES", "134217728");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES");
assert_eq!(cfg.journal_size_limit_bytes, 134_217_728);
}
#[test]
#[serial]
fn pool_config_env_override_busy_timeout() {
std::env::set_var("KHIVE_BUSY_TIMEOUT_SECS", "60");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_BUSY_TIMEOUT_SECS");
assert_eq!(cfg.busy_timeout, Duration::from_secs(60));
}
#[test]
#[serial]
fn pool_config_env_override_checkout_timeout() {
std::env::set_var("KHIVE_CHECKOUT_TIMEOUT_SECS", "10");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_CHECKOUT_TIMEOUT_SECS");
assert_eq!(cfg.checkout_timeout, Duration::from_secs(10));
}
#[test]
#[serial]
fn pool_config_write_queue_defaults_off() {
let cfg = PoolConfig::default();
assert!(!cfg.write_queue_enabled);
assert_eq!(cfg.write_queue_capacity, DEFAULT_WRITE_QUEUE_CAPACITY);
}
#[test]
#[serial]
fn pool_config_env_override_write_queue_enabled() {
std::env::set_var("KHIVE_WRITE_QUEUE", "1");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_WRITE_QUEUE");
assert!(cfg.write_queue_enabled);
}
#[test]
#[serial]
fn pool_config_env_override_write_queue_enabled_accepts_true_case_insensitive() {
std::env::set_var("KHIVE_WRITE_QUEUE", "True");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_WRITE_QUEUE");
assert!(cfg.write_queue_enabled);
}
#[test]
#[serial]
fn pool_config_env_override_write_queue_capacity() {
std::env::set_var("KHIVE_WRITE_QUEUE_CAPACITY", "64");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_WRITE_QUEUE_CAPACITY");
assert_eq!(cfg.write_queue_capacity, 64);
}
#[test]
#[serial]
fn pool_config_env_invalid_write_queue_capacity_falls_back_to_default() {
std::env::set_var("KHIVE_WRITE_QUEUE_CAPACITY", "0");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_WRITE_QUEUE_CAPACITY");
assert_eq!(cfg.write_queue_capacity, DEFAULT_WRITE_QUEUE_CAPACITY);
}
#[test]
#[serial]
fn pool_config_env_invalid_falls_back_to_default() {
std::env::set_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES", "not_a_number");
std::env::set_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES", "");
let cfg = PoolConfig::default();
std::env::remove_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES");
std::env::remove_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES");
assert_eq!(
cfg.wal_autocheckpoint_pages,
DEFAULT_WAL_AUTOCHECKPOINT_PAGES
);
assert_eq!(
cfg.journal_size_limit_bytes,
DEFAULT_JOURNAL_SIZE_LIMIT_BYTES
);
}
#[test]
fn file_backed_pool_opens_successfully() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("test_pool.db");
let cfg = PoolConfig {
path: Some(path.clone()),
..PoolConfig::default()
};
let pool = ConnectionPool::new(cfg).expect("file-backed pool should open");
assert!(path.exists());
assert!(pool.max_readers() > 0);
}
#[test]
fn in_memory_pool_degrades_to_single_connection() {
let cfg = PoolConfig {
path: None,
..PoolConfig::default()
};
let pool = ConnectionPool::new(cfg).expect("in-memory pool should open");
assert_eq!(pool.max_readers(), 0);
}
#[test]
fn writer_checkout_and_release_works() {
let cfg = PoolConfig {
path: None,
..PoolConfig::default()
};
let pool = ConnectionPool::new(cfg).unwrap();
{
let _writer = pool.writer().expect("writer checkout should succeed");
}
let _writer2 = pool
.writer()
.expect("second writer checkout should succeed");
}
#[test]
#[serial(tx_registry)]
fn writer_guard_transaction_registers_during_closure_only() {
let cfg = PoolConfig {
path: None,
..PoolConfig::default()
};
let pool = ConnectionPool::new(cfg).unwrap();
let guard = pool.writer().unwrap();
let mut seen_during_closure = false;
let result: Result<(), SqliteError> = guard.transaction(|_conn| {
seen_during_closure = khive_storage::tx_registry::snapshot()
.iter()
.any(|(_, label)| label.as_deref() == Some("writer_guard_tx"));
Ok(())
});
result.expect("transaction should commit");
assert!(
seen_during_closure,
"expected a writer_guard_tx entry visible inside the closure"
);
assert!(
!khive_storage::tx_registry::snapshot()
.iter()
.any(|(_, label)| label.as_deref() == Some("writer_guard_tx")),
"expected the entry to be gone after the transaction completes"
);
}
#[test]
fn writer_task_handle_fails_loud_without_tokio_runtime() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("writer_task_no_runtime.db");
let cfg = PoolConfig {
path: Some(path),
write_queue_enabled: true,
..PoolConfig::default()
};
let pool = ConnectionPool::new(cfg).expect("file-backed pool should open");
let result = pool.writer_task_handle();
assert!(
matches!(result, Err(StorageError::WriterTaskNoRuntime)),
"expected Err(StorageError::WriterTaskNoRuntime) outside a Tokio \
runtime, got {result:?}"
);
assert_eq!(
pool.writer_task_spawn_count(),
0,
"the guard must reject before ever attempting tokio::spawn"
);
}
#[test]
#[serial(pool_cwd)]
fn mint_db_identity_alias_convergence() {
let dir = tempfile::tempdir().unwrap();
let real_dir = dir.path().join("real");
fs::create_dir(&real_dir).unwrap();
let db_path = real_dir.join("khive.db");
fs::write(&db_path, b"").unwrap();
let dir_symlink = dir.path().join("dir_link");
let file_symlink = dir.path().join("file_link.db");
#[cfg(unix)]
{
std::os::unix::fs::symlink(&real_dir, &dir_symlink).unwrap();
std::os::unix::fs::symlink(&db_path, &file_symlink).unwrap();
}
let (via_real, canonical_real) = mint_db_identity(&db_path).unwrap();
let relative_result = {
let _cwd = CwdGuard::enter(&real_dir);
mint_db_identity(&PathBuf::from("khive.db"))
};
let (via_relative, canonical_relative) = relative_result.unwrap();
assert_eq!(canonical_real, canonical_relative);
assert_eq!(via_real, via_relative);
#[cfg(unix)]
{
let (via_dir_symlink, canonical_dir_symlink) =
mint_db_identity(&dir_symlink.join("khive.db")).unwrap();
assert_eq!(canonical_real, canonical_dir_symlink);
assert_eq!(via_real, via_dir_symlink);
let (via_file_symlink, canonical_file_symlink) =
mint_db_identity(&file_symlink).unwrap();
assert_eq!(canonical_real, canonical_file_symlink);
assert_eq!(via_real, via_file_symlink);
}
let bare_name_result = {
let _cwd = CwdGuard::enter(&real_dir);
mint_db_identity(&PathBuf::from("khive.db"))
};
let (via_bare_name, canonical_bare_name) = bare_name_result.unwrap();
assert_eq!(canonical_real, canonical_bare_name);
assert_eq!(via_real, via_bare_name);
}
#[test]
#[serial(pool_cwd)]
fn sidecar_dir_for_alias_convergence() {
let dir = tempfile::tempdir().unwrap();
let real_dir = dir.path().join("real");
fs::create_dir(&real_dir).unwrap();
let db_path = real_dir.join("khive.db");
fs::write(&db_path, b"").unwrap();
let dir_symlink = dir.path().join("dir_link");
let file_symlink = dir.path().join("file_link.db");
#[cfg(unix)]
{
std::os::unix::fs::symlink(&real_dir, &dir_symlink).unwrap();
std::os::unix::fs::symlink(&db_path, &file_symlink).unwrap();
}
let pool_for = |path: &Path| -> Arc<ConnectionPool> {
let cfg = PoolConfig {
path: Some(path.to_path_buf()),
..PoolConfig::default()
};
Arc::new(ConnectionPool::new(cfg).expect("file-backed pool should open"))
};
let sidecar_of = |pool: &ConnectionPool| -> PathBuf {
crate::walpin::sidecar_dir_for(pool.canonical_path().expect("file-backed pool"))
};
let via_real = pool_for(&db_path);
let sidecar_real = sidecar_of(&via_real);
let via_relative = {
let _cwd = CwdGuard::enter(&real_dir);
pool_for(Path::new("khive.db"))
};
assert_eq!(
sidecar_real,
sidecar_of(&via_relative),
"a relative spelling of the same database must derive the same sidecar directory"
);
#[cfg(unix)]
{
let via_dir_symlink = pool_for(&dir_symlink.join("khive.db"));
assert_eq!(
sidecar_real,
sidecar_of(&via_dir_symlink),
"opening through a directory symlink must derive the same sidecar directory"
);
let via_file_symlink = pool_for(&file_symlink);
assert_eq!(
sidecar_real,
sidecar_of(&via_file_symlink),
"opening through a file-level symlink must derive the same sidecar directory"
);
}
let via_bare_name = {
let _cwd = CwdGuard::enter(&real_dir);
pool_for(Path::new("khive.db"))
};
assert_eq!(
sidecar_real,
sidecar_of(&via_bare_name),
"a bare file name resolved against the current directory must derive the same \
sidecar directory"
);
}
#[cfg(unix)]
#[test]
fn mint_db_identity_dangling_symlink_first_open_convergence() {
let dir = tempfile::tempdir().unwrap();
let target = dir.path().join("target.db");
let link = dir.path().join("link.db");
std::os::unix::fs::symlink(&target, &link).unwrap();
assert!(!target.exists(), "target must not exist yet (dangling)");
let (via_dangling_link, canonical_via_link) = mint_db_identity(&link).unwrap();
fs::write(&target, b"").unwrap();
let (via_target, canonical_via_target) = mint_db_identity(&target).unwrap();
assert_eq!(canonical_via_link, canonical_via_target);
assert_eq!(via_dangling_link, via_target);
}
#[test]
fn mint_db_identity_missing_parent_fails() {
let dir = tempfile::tempdir().unwrap();
let missing = dir.path().join("nonexistent_subdir").join("khive.db");
let result = mint_db_identity(&missing);
assert!(
result.is_err(),
"minting must fail when the parent directory does not exist"
);
}
#[cfg(unix)]
#[test]
fn mint_db_identity_non_utf8_path_round_trips() {
use std::ffi::OsStr;
use std::os::unix::ffi::OsStrExt;
let dir = tempfile::tempdir().unwrap();
let raw_name = OsStr::from_bytes(b"khive-\xffdb.sqlite");
let db_path = dir.path().join(raw_name);
if let Err(e) = fs::write(&db_path, b"") {
eprintln!(
"skipping mint_db_identity_non_utf8_path_round_trips: filesystem rejected a \
non-UTF-8 file name ({e}); this platform's filesystem does not support the \
case under test"
);
return;
}
let (identity, canonical) = mint_db_identity(&db_path).unwrap();
assert_eq!(canonical.file_name().unwrap(), raw_name);
let (identity_again, canonical_again) = mint_db_identity(&db_path).unwrap();
assert_eq!(identity, identity_again);
assert_eq!(canonical, canonical_again);
}
}