1use crossbeam_queue::ArrayQueue;
3use parking_lot::Mutex;
4use rusqlite::{Connection, OpenFlags};
5use std::fs;
6use std::ops::{Deref, DerefMut};
7use std::path::{Path, PathBuf};
8use std::sync::{Arc, OnceLock};
9use std::thread;
10use std::time::{Duration, Instant};
11
12use crate::error::SqliteError;
13use crate::writer_task::WriterTaskHandle;
14use khive_storage::error::StorageError;
15use khive_storage::tx_registry::{DbIdentity, TxOrigin};
16
17const CACHE_SIZE_KIB: &str = "-65536";
18const MMAP_SIZE_BYTES: &str = "1073741824";
19const DEFAULT_READER_CAP: usize = 8;
20
21const DEFAULT_WAL_AUTOCHECKPOINT_PAGES: u32 = 4000;
22const DEFAULT_JOURNAL_SIZE_LIMIT_BYTES: i64 = 67_108_864; const DEFAULT_WRITE_QUEUE_CAPACITY: usize = 256;
24
25#[derive(Clone, Debug)]
27pub struct PoolConfig {
28 pub path: Option<PathBuf>,
30 pub max_readers: usize,
32 pub wal_mode: bool,
34 pub busy_timeout: Duration,
38 pub checkout_timeout: Duration,
42 pub wal_autocheckpoint_pages: u32,
49 pub journal_size_limit_bytes: i64,
55 pub read_only: bool,
63 pub write_queue_enabled: bool,
75 pub write_queue_capacity: usize,
80}
81
82impl Default for PoolConfig {
83 fn default() -> Self {
84 Self {
85 path: None,
86 max_readers: std::thread::available_parallelism()
87 .map(|n| n.get())
88 .unwrap_or(1)
89 .clamp(1, DEFAULT_READER_CAP),
90 wal_mode: true,
91 busy_timeout: Duration::from_secs(
92 std::env::var("KHIVE_BUSY_TIMEOUT_SECS")
93 .ok()
94 .and_then(|v| v.parse::<u64>().ok())
95 .unwrap_or(30),
96 ),
97 checkout_timeout: Duration::from_secs(
98 std::env::var("KHIVE_CHECKOUT_TIMEOUT_SECS")
99 .ok()
100 .and_then(|v| v.parse::<u64>().ok())
101 .unwrap_or(5),
102 ),
103 wal_autocheckpoint_pages: std::env::var("KHIVE_WAL_AUTOCHECKPOINT_PAGES")
104 .ok()
105 .and_then(|v| v.parse::<u32>().ok())
106 .unwrap_or(DEFAULT_WAL_AUTOCHECKPOINT_PAGES),
107 journal_size_limit_bytes: std::env::var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES")
108 .ok()
109 .and_then(|v| v.parse::<i64>().ok())
110 .unwrap_or(DEFAULT_JOURNAL_SIZE_LIMIT_BYTES),
111 read_only: false,
112 write_queue_enabled: std::env::var("KHIVE_WRITE_QUEUE")
113 .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
114 .unwrap_or(false),
115 write_queue_capacity: std::env::var("KHIVE_WRITE_QUEUE_CAPACITY")
116 .ok()
117 .and_then(|v| v.parse::<usize>().ok())
118 .filter(|&n| n > 0)
119 .unwrap_or(DEFAULT_WRITE_QUEUE_CAPACITY),
120 }
121 }
122}
123
124pub struct ConnectionPool {
135 writer: Arc<Mutex<Connection>>,
136 readers: ArrayQueue<Connection>,
137 max_readers: usize,
138 config: PoolConfig,
139 writer_task: OnceLock<Option<WriterTaskHandle>>,
144 origin: TxOrigin,
150 identity_path: Option<PathBuf>,
156 #[cfg(test)]
162 writer_task_spawn_count: std::sync::atomic::AtomicUsize,
163}
164
165enum ReaderLease<'pool> {
166 Pooled(Connection),
167 Shared(parking_lot::MutexGuard<'pool, Connection>),
168}
169
170pub struct ReaderGuard<'pool> {
173 lease: Option<ReaderLease<'pool>>,
174 pool: &'pool ConnectionPool,
175}
176
177impl<'pool> ReaderGuard<'pool> {
178 pub fn conn(&self) -> &Connection {
180 match self
181 .lease
182 .as_ref()
183 .expect("reader guard missing connection")
184 {
185 ReaderLease::Pooled(conn) => conn,
186 ReaderLease::Shared(guard) => guard,
187 }
188 }
189}
190
191impl<'pool> Deref for ReaderGuard<'pool> {
192 type Target = Connection;
193
194 fn deref(&self) -> &Self::Target {
195 self.conn()
196 }
197}
198
199impl<'pool> Drop for ReaderGuard<'pool> {
200 fn drop(&mut self) {
201 let Some(lease) = self.lease.take() else {
202 return;
203 };
204
205 match lease {
206 ReaderLease::Pooled(conn) => self.pool.return_reader(conn),
207 ReaderLease::Shared(_guard) => {}
208 }
209 }
210}
211
212pub struct WriterGuard<'pool> {
215 guard: parking_lot::MutexGuard<'pool, Connection>,
216 origin: TxOrigin,
220}
221
222impl<'pool> WriterGuard<'pool> {
223 pub fn conn(&self) -> &Connection {
225 &self.guard
226 }
227
228 pub fn conn_mut(&mut self) -> &mut Connection {
230 &mut self.guard
231 }
232
233 pub fn transaction<F, R>(&self, f: F) -> Result<R, SqliteError>
236 where
237 F: FnOnce(&Connection) -> Result<R, SqliteError>,
238 {
239 self.guard.execute_batch("BEGIN IMMEDIATE")?;
240 let _tx_handle = khive_storage::tx_registry::register_scoped(
241 Some("writer_guard_tx".to_string()),
242 self.origin.clone(),
243 );
244
245 match f(&self.guard) {
246 Ok(result) => {
247 if let Err(err) = self.guard.execute_batch("COMMIT") {
248 let _ = self.guard.execute_batch("ROLLBACK");
249 return Err(err.into());
250 }
251 Ok(result)
252 }
253 Err(err) => {
254 let _ = self.guard.execute_batch("ROLLBACK");
255 Err(err)
256 }
257 }
258 }
259}
260
261impl<'pool> Deref for WriterGuard<'pool> {
262 type Target = Connection;
263
264 fn deref(&self) -> &Self::Target {
265 self.conn()
266 }
267}
268
269impl<'pool> DerefMut for WriterGuard<'pool> {
270 fn deref_mut(&mut self) -> &mut Self::Target {
271 self.conn_mut()
272 }
273}
274
275impl ConnectionPool {
276 pub fn new(config: PoolConfig) -> Result<Self, SqliteError> {
284 let writer = open_writer_connection(&config)?;
285 let wal_enabled = configure_writer_connection(&writer, &config)?;
286 let max_readers = effective_reader_count(&config, wal_enabled);
287
288 let readers = ArrayQueue::new(max_readers.max(1));
289
290 let (origin, identity_path) = match config.path.as_ref() {
291 Some(path) => {
292 let (identity, canonical) = mint_db_identity(path)?;
293 (TxOrigin::Database(identity), Some(canonical))
294 }
295 None => (TxOrigin::Memory, None),
296 };
297
298 let pool = Self {
299 writer: Arc::new(Mutex::new(writer)),
300 readers,
301 max_readers,
302 config,
303 writer_task: OnceLock::new(),
304 origin,
305 identity_path,
306 #[cfg(test)]
307 writer_task_spawn_count: std::sync::atomic::AtomicUsize::new(0),
308 };
309
310 for _ in 0..pool.max_readers {
311 let conn = pool.open_reader_connection()?;
312 pool.readers
313 .push(conn)
314 .expect("reader queue must have capacity during pool initialization");
315 }
316
317 Ok(pool)
318 }
319
320 pub fn reader(&self) -> Result<ReaderGuard<'_>, SqliteError> {
332 if self.max_readers == 0 {
333 return Ok(ReaderGuard {
334 lease: Some(ReaderLease::Shared(self.writer.lock())),
335 pool: self,
336 });
337 }
338
339 let started = Instant::now();
340 let mut attempt = 0u32;
341
342 loop {
343 if let Some(conn) = self.readers.pop() {
344 return Ok(ReaderGuard {
345 lease: Some(ReaderLease::Pooled(conn)),
346 pool: self,
347 });
348 }
349
350 if started.elapsed() >= self.config.checkout_timeout {
351 return Err(pool_exhausted_error(
352 self.config.checkout_timeout,
353 self.max_readers,
354 ));
355 }
356
357 match attempt {
358 0..=7 => {
359 let spins = 1usize << attempt;
360 for _ in 0..spins {
361 std::hint::spin_loop();
362 }
363 }
364 8..=15 => thread::yield_now(),
365 _ => {
366 let remaining = self
367 .config
368 .checkout_timeout
369 .saturating_sub(started.elapsed());
370 let sleep = Duration::from_micros(50 * (1u64 << (attempt - 16).min(6)));
371 thread::sleep(sleep.min(remaining).min(Duration::from_millis(2)));
372 }
373 }
374
375 attempt = attempt.saturating_add(1);
376 }
377 }
378
379 pub fn writer(&self) -> Result<WriterGuard<'_>, SqliteError> {
384 let guard = self
385 .writer
386 .try_lock_for(self.config.checkout_timeout)
387 .ok_or_else(|| {
388 SqliteError::InvalidData(format!(
389 "timed out after {:?} waiting for sqlite writer connection",
390 self.config.checkout_timeout
391 ))
392 })?;
393 Ok(WriterGuard {
394 guard,
395 origin: self.origin(),
396 })
397 }
398
399 pub fn try_writer(&self) -> Result<WriterGuard<'_>, SqliteError> {
404 self.writer()
405 }
406
407 pub fn try_writer_nowait(&self) -> Result<WriterGuard<'_>, SqliteError> {
416 let guard = self.writer.try_lock().ok_or_else(|| {
417 SqliteError::InvalidData(
418 "writer connection busy (checkpoint skipped this tick)".to_string(),
419 )
420 })?;
421 Ok(WriterGuard {
422 guard,
423 origin: self.origin(),
424 })
425 }
426
427 pub fn available_readers(&self) -> usize {
429 self.readers.len()
430 }
431
432 pub fn max_readers(&self) -> usize {
434 self.max_readers
435 }
436
437 pub fn config(&self) -> &PoolConfig {
439 &self.config
440 }
441
442 pub fn origin(&self) -> TxOrigin {
448 self.origin.clone()
449 }
450
451 pub fn canonical_path(&self) -> Option<&Path> {
457 self.identity_path.as_deref()
458 }
459
460 pub fn writer_task_handle(&self) -> Result<Option<WriterTaskHandle>, StorageError> {
484 if !self.config.write_queue_enabled {
485 return Ok(None);
486 }
487 if let Some(existing) = self.writer_task.get() {
490 return Ok(existing.clone());
491 }
492 if tokio::runtime::Handle::try_current().is_err() {
496 return Err(StorageError::WriterTaskNoRuntime);
497 }
498 Ok(self
499 .writer_task
500 .get_or_init(|| {
501 #[cfg(test)]
502 self.writer_task_spawn_count
503 .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
504
505 match crate::writer_task::spawn(self, self.config.write_queue_capacity) {
506 Ok(handle) => Some(handle),
507 Err(e) => {
508 tracing::warn!(
509 error = %e,
510 "KHIVE_WRITE_QUEUE=1 but the writer task failed to spawn; \
511 writes fall back to the pool-mutex path"
512 );
513 None
514 }
515 }
516 })
517 .clone())
518 }
519
520 #[cfg(test)]
525 pub(crate) fn writer_task_spawn_count(&self) -> usize {
526 self.writer_task_spawn_count
527 .load(std::sync::atomic::Ordering::SeqCst)
528 }
529
530 pub fn legacy_conn(&self) -> Arc<Mutex<Connection>> {
535 Arc::clone(&self.writer)
536 }
537
538 fn open_reader_connection(&self) -> Result<Connection, SqliteError> {
539 let path = self
540 .config
541 .path
542 .as_ref()
543 .expect("reader connections require a file-backed database");
544 open_reader_connection(path, &self.config)
545 }
546
547 pub fn open_standalone_writer(&self) -> Result<Connection, SqliteError> {
557 let path = self.config.path.as_ref().ok_or_else(|| {
558 SqliteError::InvalidData(
559 "in-memory databases do not support standalone connections".to_string(),
560 )
561 })?;
562
563 if self.config.read_only {
564 return Err(SqliteError::InvalidData(
565 "database is read-only: standalone write connections are not permitted".to_string(),
566 ));
567 }
568
569 let conn = Connection::open_with_flags(
570 path,
571 OpenFlags::SQLITE_OPEN_READ_WRITE
572 | OpenFlags::SQLITE_OPEN_NO_MUTEX
573 | OpenFlags::SQLITE_OPEN_URI,
574 )?;
575 conn.busy_timeout(self.config.busy_timeout)?;
576 conn.pragma_update(None, "foreign_keys", "ON")?;
577 conn.pragma_update(None, "synchronous", "NORMAL")?;
578 Ok(conn)
579 }
580
581 pub fn open_standalone_reader(&self) -> Result<Connection, SqliteError> {
586 let path = self.config.path.as_ref().ok_or_else(|| {
587 SqliteError::InvalidData(
588 "in-memory databases do not support standalone connections".to_string(),
589 )
590 })?;
591
592 let conn = Connection::open_with_flags(
593 path,
594 OpenFlags::SQLITE_OPEN_READ_ONLY
595 | OpenFlags::SQLITE_OPEN_NO_MUTEX
596 | OpenFlags::SQLITE_OPEN_URI,
597 )?;
598 conn.busy_timeout(self.config.busy_timeout)?;
599 conn.pragma_update(None, "foreign_keys", "ON")?;
600 conn.pragma_update(None, "synchronous", "NORMAL")?;
601 Ok(conn)
602 }
603
604 fn return_reader(&self, conn: Connection) {
605 if self.max_readers == 0 {
606 return;
607 }
608
609 let conn = if reset_reader_connection(&conn) && reader_connection_is_healthy(&conn) {
610 Some(conn)
611 } else {
612 close_connection_quietly(conn);
613 self.open_reader_connection().ok()
614 };
615
616 if let Some(conn) = conn {
617 if let Err(conn) = self.readers.push(conn) {
618 eprintln!(
619 "[sqlite-pool] reader pool queue full, discarding replacement connection"
620 );
621 close_connection_quietly(conn);
622 }
623 }
624 }
625}
626
627const MAX_SYMLINK_DEPTH: u32 = 40;
632
633fn mint_db_identity(configured_path: &Path) -> Result<(DbIdentity, PathBuf), SqliteError> {
666 let absolute = if configured_path.is_absolute() {
667 configured_path.to_path_buf()
668 } else {
669 let cwd = std::env::current_dir().map_err(|e| {
670 SqliteError::InvalidData(format!(
671 "cannot mint database identity for {configured_path:?}: failed to resolve the \
672 process current directory: {e}"
673 ))
674 })?;
675 cwd.join(configured_path)
676 };
677
678 if absolute.exists() {
679 let canonical = absolute.canonicalize().map_err(|e| {
680 SqliteError::InvalidData(format!(
681 "cannot mint database identity: failed to canonicalize existing path \
682 {absolute:?}: {e}"
683 ))
684 })?;
685 return Ok((
686 DbIdentity::new(canonical.clone().into_os_string()),
687 canonical,
688 ));
689 }
690
691 let resolved_target = resolve_symlink_chain(&absolute)?;
692 let parent = resolved_target.parent().ok_or_else(|| {
693 SqliteError::InvalidData(format!(
694 "cannot mint database identity for {resolved_target:?}: path has no parent \
695 directory"
696 ))
697 })?;
698 let file_name = resolved_target.file_name().ok_or_else(|| {
699 SqliteError::InvalidData(format!(
700 "cannot mint database identity for {resolved_target:?}: path has no file name"
701 ))
702 })?;
703 let canonical_parent = parent.canonicalize().map_err(|e| {
704 SqliteError::InvalidData(format!(
705 "cannot mint database identity: parent directory {parent:?} of first-open path \
706 {resolved_target:?} does not exist or is inaccessible: {e}"
707 ))
708 })?;
709 let mut identity_path = canonical_parent;
710 identity_path.push(file_name);
711 Ok((
712 DbIdentity::new(identity_path.clone().into_os_string()),
713 identity_path,
714 ))
715}
716
717fn resolve_symlink_chain(path: &Path) -> Result<PathBuf, SqliteError> {
723 let mut current = path.to_path_buf();
724 for _ in 0..MAX_SYMLINK_DEPTH {
725 match fs::symlink_metadata(¤t) {
726 Ok(meta) if meta.file_type().is_symlink() => {
727 let target = fs::read_link(¤t).map_err(|e| {
728 SqliteError::InvalidData(format!(
729 "cannot mint database identity: failed to read symlink {current:?}: {e}"
730 ))
731 })?;
732 current = if target.is_absolute() {
733 target
734 } else {
735 match current.parent() {
736 Some(parent) => parent.join(&target),
737 None => target,
738 }
739 };
740 }
741 _ => return Ok(current),
742 }
743 }
744 Err(SqliteError::InvalidData(format!(
745 "cannot mint database identity for {path:?}: symlink chain exceeds \
746 {MAX_SYMLINK_DEPTH} levels"
747 )))
748}
749
750fn effective_reader_count(config: &PoolConfig, wal_enabled: bool) -> usize {
751 if config.path.is_some() && config.wal_mode && wal_enabled {
752 config.max_readers
753 } else {
754 0
755 }
756}
757
758fn open_writer_connection(config: &PoolConfig) -> Result<Connection, SqliteError> {
759 match config.path.as_ref() {
760 Some(path) => {
761 let flags = if config.read_only {
762 writer_read_only_open_flags()
763 } else {
764 writer_open_flags()
765 };
766 Connection::open_with_flags(path, flags).map_err(Into::into)
767 }
768 None => Connection::open_in_memory().map_err(Into::into),
769 }
770}
771
772fn open_reader_connection(path: &Path, config: &PoolConfig) -> Result<Connection, SqliteError> {
773 let conn = Connection::open_with_flags(path, reader_open_flags())?;
774 configure_reader_connection(&conn, config)?;
775 Ok(conn)
776}
777
778fn writer_open_flags() -> OpenFlags {
779 OpenFlags::SQLITE_OPEN_READ_WRITE
780 | OpenFlags::SQLITE_OPEN_CREATE
781 | OpenFlags::SQLITE_OPEN_URI
782 | OpenFlags::SQLITE_OPEN_NO_MUTEX
783}
784
785fn writer_read_only_open_flags() -> OpenFlags {
788 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX
789}
790
791fn reader_open_flags() -> OpenFlags {
792 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX
793}
794
795fn configure_writer_connection(
796 conn: &Connection,
797 config: &PoolConfig,
798) -> Result<bool, SqliteError> {
799 if config.read_only {
800 conn.pragma_update(None, "foreign_keys", "ON")?;
804 conn.busy_timeout(config.busy_timeout)?;
805 conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
806 conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
807 conn.pragma_update(None, "temp_store", "MEMORY")?;
808 conn.pragma_update(None, "query_only", "ON")?;
809
810 let wal_enabled =
811 config.wal_mode && current_journal_mode(conn)?.eq_ignore_ascii_case("wal");
812 return Ok(wal_enabled);
813 }
814
815 let wants_wal = config.path.is_some() && config.wal_mode;
816
817 if wants_wal {
818 conn.pragma_update(None, "journal_mode", "WAL")?;
819 }
820
821 conn.pragma_update(None, "synchronous", "NORMAL")?;
822 conn.pragma_update(None, "foreign_keys", "ON")?;
823 conn.busy_timeout(config.busy_timeout)?;
824 conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
825 conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
826 conn.pragma_update(None, "temp_store", "MEMORY")?;
827
828 let wal_enabled = wants_wal && current_journal_mode(conn)?.eq_ignore_ascii_case("wal");
829
830 if wal_enabled {
831 conn.pragma_update(None, "wal_autocheckpoint", config.wal_autocheckpoint_pages)?;
832 conn.pragma_update(None, "journal_size_limit", config.journal_size_limit_bytes)?;
833 }
834
835 Ok(wal_enabled)
836}
837
838fn configure_reader_connection(conn: &Connection, config: &PoolConfig) -> Result<(), SqliteError> {
839 conn.pragma_update(None, "foreign_keys", "ON")?;
840 conn.busy_timeout(config.busy_timeout)?;
841 conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
842 conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
843 conn.pragma_update(None, "temp_store", "MEMORY")?;
844 Ok(())
845}
846
847fn current_journal_mode(conn: &Connection) -> Result<String, SqliteError> {
848 conn.pragma_query_value(None, "journal_mode", |row| row.get::<_, String>(0))
849 .map(|mode| mode.to_ascii_lowercase())
850 .map_err(Into::into)
851}
852
853fn reset_reader_connection(conn: &Connection) -> bool {
854 if conn.is_autocommit() {
855 return true;
856 }
857
858 match conn.execute_batch("ROLLBACK") {
859 Ok(()) => conn.is_autocommit(),
860 Err(rusqlite::Error::SqliteFailure(err, _)) => {
861 if matches!(
862 err.code,
863 rusqlite::ErrorCode::CannotOpen
864 | rusqlite::ErrorCode::DatabaseCorrupt
865 | rusqlite::ErrorCode::NotADatabase
866 | rusqlite::ErrorCode::DiskFull
867 ) {
868 return false;
869 }
870 conn.is_autocommit()
871 }
872 Err(_) => false,
873 }
874}
875
876fn reader_connection_is_healthy(conn: &Connection) -> bool {
877 match conn.query_row("SELECT 1", [], |row| row.get::<_, i64>(0)) {
878 Ok(_) => true,
879 Err(rusqlite::Error::SqliteFailure(err, _)) => !matches!(
880 err.code,
881 rusqlite::ErrorCode::CannotOpen
882 | rusqlite::ErrorCode::NotADatabase
883 | rusqlite::ErrorCode::DatabaseCorrupt
884 | rusqlite::ErrorCode::PermissionDenied
885 | rusqlite::ErrorCode::SystemIoFailure
886 ),
887 Err(_) => true,
888 }
889}
890
891fn close_connection_quietly(conn: Connection) {
892 match conn.close() {
893 Ok(()) => {}
894 Err((conn, _)) => drop(conn),
895 }
896}
897
898fn pool_exhausted_error(timeout: Duration, max_readers: usize) -> SqliteError {
899 rusqlite::Error::SqliteFailure(
900 rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
901 Some(format!(
902 "Pool exhausted: no reader available after {timeout:?} (max_readers={max_readers})"
903 )),
904 )
905 .into()
906}
907
908#[cfg(test)]
909mod tests {
910 use super::*;
911 use serial_test::serial;
912
913 struct CwdGuard {
918 original: PathBuf,
919 }
920
921 impl CwdGuard {
922 fn enter(dir: &Path) -> Self {
923 let original = std::env::current_dir().unwrap();
924 std::env::set_current_dir(dir).unwrap();
925 Self { original }
926 }
927 }
928
929 impl Drop for CwdGuard {
930 fn drop(&mut self) {
931 let _ = std::env::set_current_dir(&self.original);
932 }
933 }
934
935 #[test]
936 #[serial]
937 fn pool_config_default_values_match_constants() {
938 let cfg = PoolConfig::default();
940 assert_eq!(
941 cfg.wal_autocheckpoint_pages,
942 DEFAULT_WAL_AUTOCHECKPOINT_PAGES
943 );
944 assert_eq!(
945 cfg.journal_size_limit_bytes,
946 DEFAULT_JOURNAL_SIZE_LIMIT_BYTES
947 );
948 assert_eq!(cfg.busy_timeout, Duration::from_secs(30));
949 assert_eq!(cfg.checkout_timeout, Duration::from_secs(5));
950 }
951
952 #[test]
953 #[serial]
954 fn pool_config_env_override_wal_autocheckpoint() {
955 std::env::set_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES", "8000");
956 let cfg = PoolConfig::default();
957 std::env::remove_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES");
958 assert_eq!(cfg.wal_autocheckpoint_pages, 8000);
959 }
960
961 #[test]
962 #[serial]
963 fn pool_config_env_override_journal_size_limit() {
964 std::env::set_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES", "134217728");
965 let cfg = PoolConfig::default();
966 std::env::remove_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES");
967 assert_eq!(cfg.journal_size_limit_bytes, 134_217_728);
968 }
969
970 #[test]
971 #[serial]
972 fn pool_config_env_override_busy_timeout() {
973 std::env::set_var("KHIVE_BUSY_TIMEOUT_SECS", "60");
974 let cfg = PoolConfig::default();
975 std::env::remove_var("KHIVE_BUSY_TIMEOUT_SECS");
976 assert_eq!(cfg.busy_timeout, Duration::from_secs(60));
977 }
978
979 #[test]
980 #[serial]
981 fn pool_config_env_override_checkout_timeout() {
982 std::env::set_var("KHIVE_CHECKOUT_TIMEOUT_SECS", "10");
983 let cfg = PoolConfig::default();
984 std::env::remove_var("KHIVE_CHECKOUT_TIMEOUT_SECS");
985 assert_eq!(cfg.checkout_timeout, Duration::from_secs(10));
986 }
987
988 #[test]
989 #[serial]
990 fn pool_config_write_queue_defaults_off() {
991 let cfg = PoolConfig::default();
992 assert!(!cfg.write_queue_enabled);
993 assert_eq!(cfg.write_queue_capacity, DEFAULT_WRITE_QUEUE_CAPACITY);
994 }
995
996 #[test]
997 #[serial]
998 fn pool_config_env_override_write_queue_enabled() {
999 std::env::set_var("KHIVE_WRITE_QUEUE", "1");
1000 let cfg = PoolConfig::default();
1001 std::env::remove_var("KHIVE_WRITE_QUEUE");
1002 assert!(cfg.write_queue_enabled);
1003 }
1004
1005 #[test]
1006 #[serial]
1007 fn pool_config_env_override_write_queue_enabled_accepts_true_case_insensitive() {
1008 std::env::set_var("KHIVE_WRITE_QUEUE", "True");
1009 let cfg = PoolConfig::default();
1010 std::env::remove_var("KHIVE_WRITE_QUEUE");
1011 assert!(cfg.write_queue_enabled);
1012 }
1013
1014 #[test]
1015 #[serial]
1016 fn pool_config_env_override_write_queue_capacity() {
1017 std::env::set_var("KHIVE_WRITE_QUEUE_CAPACITY", "64");
1018 let cfg = PoolConfig::default();
1019 std::env::remove_var("KHIVE_WRITE_QUEUE_CAPACITY");
1020 assert_eq!(cfg.write_queue_capacity, 64);
1021 }
1022
1023 #[test]
1024 #[serial]
1025 fn pool_config_env_invalid_write_queue_capacity_falls_back_to_default() {
1026 std::env::set_var("KHIVE_WRITE_QUEUE_CAPACITY", "0");
1027 let cfg = PoolConfig::default();
1028 std::env::remove_var("KHIVE_WRITE_QUEUE_CAPACITY");
1029 assert_eq!(cfg.write_queue_capacity, DEFAULT_WRITE_QUEUE_CAPACITY);
1030 }
1031
1032 #[test]
1033 #[serial]
1034 fn pool_config_env_invalid_falls_back_to_default() {
1035 std::env::set_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES", "not_a_number");
1036 std::env::set_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES", "");
1037 let cfg = PoolConfig::default();
1038 std::env::remove_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES");
1039 std::env::remove_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES");
1040 assert_eq!(
1041 cfg.wal_autocheckpoint_pages,
1042 DEFAULT_WAL_AUTOCHECKPOINT_PAGES
1043 );
1044 assert_eq!(
1045 cfg.journal_size_limit_bytes,
1046 DEFAULT_JOURNAL_SIZE_LIMIT_BYTES
1047 );
1048 }
1049
1050 #[test]
1051 fn file_backed_pool_opens_successfully() {
1052 let dir = tempfile::tempdir().unwrap();
1053 let path = dir.path().join("test_pool.db");
1054 let cfg = PoolConfig {
1055 path: Some(path.clone()),
1056 ..PoolConfig::default()
1057 };
1058 let pool = ConnectionPool::new(cfg).expect("file-backed pool should open");
1059 assert!(path.exists());
1060 assert!(pool.max_readers() > 0);
1061 }
1062
1063 #[test]
1064 fn in_memory_pool_degrades_to_single_connection() {
1065 let cfg = PoolConfig {
1066 path: None,
1067 ..PoolConfig::default()
1068 };
1069 let pool = ConnectionPool::new(cfg).expect("in-memory pool should open");
1070 assert_eq!(pool.max_readers(), 0);
1071 }
1072
1073 #[test]
1074 fn writer_checkout_and_release_works() {
1075 let cfg = PoolConfig {
1076 path: None,
1077 ..PoolConfig::default()
1078 };
1079 let pool = ConnectionPool::new(cfg).unwrap();
1080 {
1081 let _writer = pool.writer().expect("writer checkout should succeed");
1082 }
1083 let _writer2 = pool
1085 .writer()
1086 .expect("second writer checkout should succeed");
1087 }
1088
1089 #[test]
1093 #[serial(tx_registry)]
1094 fn writer_guard_transaction_registers_during_closure_only() {
1095 let cfg = PoolConfig {
1096 path: None,
1097 ..PoolConfig::default()
1098 };
1099 let pool = ConnectionPool::new(cfg).unwrap();
1100 let guard = pool.writer().unwrap();
1101
1102 let mut seen_during_closure = false;
1103 let result: Result<(), SqliteError> = guard.transaction(|_conn| {
1104 seen_during_closure = khive_storage::tx_registry::snapshot()
1105 .iter()
1106 .any(|(_, label)| label.as_deref() == Some("writer_guard_tx"));
1107 Ok(())
1108 });
1109 result.expect("transaction should commit");
1110
1111 assert!(
1112 seen_during_closure,
1113 "expected a writer_guard_tx entry visible inside the closure"
1114 );
1115 assert!(
1116 !khive_storage::tx_registry::snapshot()
1117 .iter()
1118 .any(|(_, label)| label.as_deref() == Some("writer_guard_tx")),
1119 "expected the entry to be gone after the transaction completes"
1120 );
1121 }
1122
1123 #[test]
1127 fn writer_task_handle_fails_loud_without_tokio_runtime() {
1128 let dir = tempfile::tempdir().unwrap();
1129 let path = dir.path().join("writer_task_no_runtime.db");
1130 let cfg = PoolConfig {
1131 path: Some(path),
1132 write_queue_enabled: true,
1133 ..PoolConfig::default()
1134 };
1135 let pool = ConnectionPool::new(cfg).expect("file-backed pool should open");
1136
1137 let result = pool.writer_task_handle();
1138
1139 assert!(
1140 matches!(result, Err(StorageError::WriterTaskNoRuntime)),
1141 "expected Err(StorageError::WriterTaskNoRuntime) outside a Tokio \
1142 runtime, got {result:?}"
1143 );
1144 assert_eq!(
1145 pool.writer_task_spawn_count(),
1146 0,
1147 "the guard must reject before ever attempting tokio::spawn"
1148 );
1149 }
1150
1151 #[test]
1156 #[serial(pool_cwd)]
1157 fn mint_db_identity_alias_convergence() {
1158 let dir = tempfile::tempdir().unwrap();
1159 let real_dir = dir.path().join("real");
1160 fs::create_dir(&real_dir).unwrap();
1161 let db_path = real_dir.join("khive.db");
1162 fs::write(&db_path, b"").unwrap();
1163
1164 let dir_symlink = dir.path().join("dir_link");
1165 let file_symlink = dir.path().join("file_link.db");
1166 #[cfg(unix)]
1167 {
1168 std::os::unix::fs::symlink(&real_dir, &dir_symlink).unwrap();
1169 std::os::unix::fs::symlink(&db_path, &file_symlink).unwrap();
1170 }
1171
1172 let (via_real, canonical_real) = mint_db_identity(&db_path).unwrap();
1173
1174 let relative_result = {
1176 let _cwd = CwdGuard::enter(&real_dir);
1177 mint_db_identity(&PathBuf::from("khive.db"))
1178 };
1179 let (via_relative, canonical_relative) = relative_result.unwrap();
1180 assert_eq!(canonical_real, canonical_relative);
1181 assert_eq!(via_real, via_relative);
1182
1183 #[cfg(unix)]
1184 {
1185 let (via_dir_symlink, canonical_dir_symlink) =
1186 mint_db_identity(&dir_symlink.join("khive.db")).unwrap();
1187 assert_eq!(canonical_real, canonical_dir_symlink);
1188 assert_eq!(via_real, via_dir_symlink);
1189
1190 let (via_file_symlink, canonical_file_symlink) =
1191 mint_db_identity(&file_symlink).unwrap();
1192 assert_eq!(canonical_real, canonical_file_symlink);
1193 assert_eq!(via_real, via_file_symlink);
1194 }
1195
1196 let bare_name_result = {
1198 let _cwd = CwdGuard::enter(&real_dir);
1199 mint_db_identity(&PathBuf::from("khive.db"))
1200 };
1201 let (via_bare_name, canonical_bare_name) = bare_name_result.unwrap();
1202 assert_eq!(canonical_real, canonical_bare_name);
1203 assert_eq!(via_real, via_bare_name);
1204 }
1205
1206 #[test]
1218 #[serial(pool_cwd)]
1219 fn sidecar_dir_for_alias_convergence() {
1220 let dir = tempfile::tempdir().unwrap();
1221 let real_dir = dir.path().join("real");
1222 fs::create_dir(&real_dir).unwrap();
1223 let db_path = real_dir.join("khive.db");
1224 fs::write(&db_path, b"").unwrap();
1225
1226 let dir_symlink = dir.path().join("dir_link");
1227 let file_symlink = dir.path().join("file_link.db");
1228 #[cfg(unix)]
1229 {
1230 std::os::unix::fs::symlink(&real_dir, &dir_symlink).unwrap();
1231 std::os::unix::fs::symlink(&db_path, &file_symlink).unwrap();
1232 }
1233
1234 let pool_for = |path: &Path| -> Arc<ConnectionPool> {
1235 let cfg = PoolConfig {
1236 path: Some(path.to_path_buf()),
1237 ..PoolConfig::default()
1238 };
1239 Arc::new(ConnectionPool::new(cfg).expect("file-backed pool should open"))
1240 };
1241 let sidecar_of = |pool: &ConnectionPool| -> PathBuf {
1242 crate::walpin::sidecar_dir_for(pool.canonical_path().expect("file-backed pool"))
1243 };
1244
1245 let via_real = pool_for(&db_path);
1246 let sidecar_real = sidecar_of(&via_real);
1247
1248 let via_relative = {
1249 let _cwd = CwdGuard::enter(&real_dir);
1250 pool_for(Path::new("khive.db"))
1251 };
1252 assert_eq!(
1253 sidecar_real,
1254 sidecar_of(&via_relative),
1255 "a relative spelling of the same database must derive the same sidecar directory"
1256 );
1257
1258 #[cfg(unix)]
1259 {
1260 let via_dir_symlink = pool_for(&dir_symlink.join("khive.db"));
1261 assert_eq!(
1262 sidecar_real,
1263 sidecar_of(&via_dir_symlink),
1264 "opening through a directory symlink must derive the same sidecar directory"
1265 );
1266
1267 let via_file_symlink = pool_for(&file_symlink);
1268 assert_eq!(
1269 sidecar_real,
1270 sidecar_of(&via_file_symlink),
1271 "opening through a file-level symlink must derive the same sidecar directory"
1272 );
1273 }
1274
1275 let via_bare_name = {
1276 let _cwd = CwdGuard::enter(&real_dir);
1277 pool_for(Path::new("khive.db"))
1278 };
1279 assert_eq!(
1280 sidecar_real,
1281 sidecar_of(&via_bare_name),
1282 "a bare file name resolved against the current directory must derive the same \
1283 sidecar directory"
1284 );
1285 }
1286
1287 #[cfg(unix)]
1293 #[test]
1294 fn mint_db_identity_dangling_symlink_first_open_convergence() {
1295 let dir = tempfile::tempdir().unwrap();
1296 let target = dir.path().join("target.db");
1297 let link = dir.path().join("link.db");
1298 std::os::unix::fs::symlink(&target, &link).unwrap();
1299 assert!(!target.exists(), "target must not exist yet (dangling)");
1300
1301 let (via_dangling_link, canonical_via_link) = mint_db_identity(&link).unwrap();
1302
1303 fs::write(&target, b"").unwrap();
1306 let (via_target, canonical_via_target) = mint_db_identity(&target).unwrap();
1307
1308 assert_eq!(canonical_via_link, canonical_via_target);
1309 assert_eq!(via_dangling_link, via_target);
1310 }
1311
1312 #[test]
1315 fn mint_db_identity_missing_parent_fails() {
1316 let dir = tempfile::tempdir().unwrap();
1317 let missing = dir.path().join("nonexistent_subdir").join("khive.db");
1318 let result = mint_db_identity(&missing);
1319 assert!(
1320 result.is_err(),
1321 "minting must fail when the parent directory does not exist"
1322 );
1323 }
1324
1325 #[cfg(unix)]
1328 #[test]
1329 fn mint_db_identity_non_utf8_path_round_trips() {
1330 use std::ffi::OsStr;
1331 use std::os::unix::ffi::OsStrExt;
1332
1333 let dir = tempfile::tempdir().unwrap();
1334 let raw_name = OsStr::from_bytes(b"khive-\xffdb.sqlite");
1336 let db_path = dir.path().join(raw_name);
1337 if let Err(e) = fs::write(&db_path, b"") {
1342 eprintln!(
1343 "skipping mint_db_identity_non_utf8_path_round_trips: filesystem rejected a \
1344 non-UTF-8 file name ({e}); this platform's filesystem does not support the \
1345 case under test"
1346 );
1347 return;
1348 }
1349
1350 let (identity, canonical) = mint_db_identity(&db_path).unwrap();
1351 assert_eq!(canonical.file_name().unwrap(), raw_name);
1352
1353 let (identity_again, canonical_again) = mint_db_identity(&db_path).unwrap();
1354 assert_eq!(identity, identity_again);
1355 assert_eq!(canonical, canonical_again);
1356 }
1357}