From de94f8d5e0663cfcf5956b68d4b52c4a29620ced Mon Sep 17 00:00:00 2001 From: "Luke Boswell (Linux-Desktop)" Date: Wed, 22 Jul 2026 11:36:48 +1000 Subject: [PATCH] Close unused SQLite connections --- README.md | 11 +- src/sqlite.rs | 355 +++++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 340 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index ebc97b61..a5016bf8 100644 --- a/README.md +++ b/README.md @@ -44,12 +44,11 @@ These current host-level limitations may affect application design: an error. `Tcp.Stream.read_until!` and `read_line!` also have no size limit and buffer until the delimiter or EOF. [Issue #437](https://github.com/roc-lang/basic-cli/issues/437) tracks timeout and bounded-read APIs. -- SQLite caches one connection per database path for the life of the process. - Reusing a path reuses its connection, so repeated use of `:memory:` accesses - the same in-memory database. Using many distinct paths retains all of those - connections. Prepared statements are finalized after their last reference is - dropped, but the cached connections remain open. [Issue #435](https://github.com/roc-lang/basic-cli/issues/435) - tracks connection ownership and cleanup. +- SQLite keeps up to 16 recently used ordinary path connections open. A live + prepared statement keeps its connection open after cache eviction, and reused + paths continue sharing that live connection. The exact `:memory:` path is kept + open for the lifetime of its host thread, so repeated use accesses the same + in-memory database. ## Examples diff --git a/src/sqlite.rs b/src/sqlite.rs index c8177735..26dcf0ba 100644 --- a/src/sqlite.rs +++ b/src/sqlite.rs @@ -1,6 +1,7 @@ use core::mem::ManuallyDrop; use std::cell::RefCell; use std::ffi::{c_char, c_int, c_void, CStr, CString}; +use std::rc::{Rc, Weak}; use crate::roc_platform_abi::*; use crate::{roc_host, roc_u8_list_from_slice}; @@ -23,9 +24,18 @@ type NativePath = UnixBytesOrUtf8OrWindowsU16s; type NativePathTag = UnixBytesOrUtf8OrWindowsU16sTag; const SQLITE_STMT_BOX_ALIGN: usize = core::mem::align_of::(); +const MAX_CACHED_CONNECTIONS: usize = 16; + +#[cfg(test)] +static FAILED_OPEN_HANDLES_CLOSED: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + +struct SqliteConnection { + raw: *mut libsqlite3_sys::sqlite3, +} struct SqliteStatement { - connection: *mut libsqlite3_sys::sqlite3, + connection: Rc, stmt: *mut libsqlite3_sys::sqlite3_stmt, } @@ -35,6 +45,104 @@ enum SqlitePath { Windows(Vec), } +impl SqlitePath { + fn is_memory(&self) -> bool { + match self { + Self::Unix(path) => path == b":memory:", + Self::Windows(path) => { + path.as_slice() + == [ + b':' as u16, + b'm' as u16, + b'e' as u16, + b'm' as u16, + b'o' as u16, + b'r' as u16, + b'y' as u16, + b':' as u16, + ] + } + } + } +} + +struct CachedConnection { + path: SqlitePath, + connection: Weak, + keep_alive: Option>, + last_used: u64, +} + +struct ConnectionCache { + entries: Vec, + clock: u64, +} + +impl ConnectionCache { + fn get(&mut self, path: SqlitePath) -> Result, (c_int, String)> { + self.entries + .retain(|entry| entry.keep_alive.is_some() || entry.connection.strong_count() > 0); + + self.clock = self.clock.wrapping_add(1); + let last_used = self.clock; + + if let Some(entry) = self.entries.iter_mut().find(|entry| entry.path == path) { + if let Some(connection) = entry.connection.upgrade() { + entry.keep_alive = Some(Rc::clone(&connection)); + entry.last_used = last_used; + self.evict_unused_connections(); + return Ok(connection); + } + } + + let connection = Rc::new(SqliteConnection { + raw: sqlite_open_native(&path)?, + }); + self.entries.push(CachedConnection { + connection: Rc::downgrade(&connection), + keep_alive: Some(Rc::clone(&connection)), + path, + last_used, + }); + self.evict_unused_connections(); + Ok(connection) + } + + fn evict_unused_connections(&mut self) { + while self + .entries + .iter() + .filter(|entry| !entry.path.is_memory() && entry.keep_alive.is_some()) + .count() + > MAX_CACHED_CONNECTIONS + { + let Some(index) = self + .entries + .iter() + .enumerate() + .filter(|(_, entry)| !entry.path.is_memory() && entry.keep_alive.is_some()) + .min_by_key(|(_, entry)| entry.last_used) + .map(|(index, _)| index) + else { + break; + }; + self.entries[index].keep_alive = None; + } + } +} + +impl SqliteConnection { + fn as_ptr(&self) -> *mut libsqlite3_sys::sqlite3 { + self.raw + } +} + +impl Drop for SqliteConnection { + fn drop(&mut self) { + sqlite_close(self.raw, false); + } +} + impl Drop for SqliteStatement { fn drop(&mut self) { unsafe { @@ -44,9 +152,15 @@ impl Drop for SqliteStatement { } thread_local! { - // Connections are cached per database path and live until process exit. - static SQLITE_CONNECTIONS: RefCell> = - const { RefCell::new(Vec::new()) }; + // Keep a bounded set of ordinary connections warm. Weak entries preserve + // sharing with live statements after eviction. `:memory:` is deliberately + // pinned so repeated uses on this thread keep addressing the same database. + static SQLITE_CONNECTIONS: RefCell = const { + RefCell::new(ConnectionCache { + entries: Vec::new(), + clock: 0, + }) + }; } fn box_sqlite_stmt(stmt: SqliteStatement, roc_host: &RocHost) -> *mut u64 { @@ -123,7 +237,7 @@ fn sqlite_error(code: c_int, message: &str, roc_host: &RocHost) -> SqliteError { } fn sqlite_err_from_stmt(stmt: &SqliteStatement, code: c_int, roc_host: &RocHost) -> SqliteError { - let message = sqlite_errmsg(stmt.connection, code); + let message = sqlite_errmsg(stmt.connection.as_ptr(), code); sqlite_error(code, &message, roc_host) } @@ -169,6 +283,12 @@ extern "C" { fn sqlite3_open16(filename: *const c_void, pp_db: *mut *mut libsqlite3_sys::sqlite3) -> c_int; } +// libsqlite3-sys 0.33 deliberately omits this stable SQLite API from its Rust +// bindings, although the bundled SQLite library exports it. +extern "C" { + fn sqlite3_close_v2(connection: *mut libsqlite3_sys::sqlite3) -> c_int; +} + fn sqlite_open_native(path: &SqlitePath) -> Result<*mut libsqlite3_sys::sqlite3, (c_int, String)> { let mut connection: *mut libsqlite3_sys::sqlite3 = core::ptr::null_mut(); @@ -195,6 +315,7 @@ fn sqlite_open_native(path: &SqlitePath) -> Result<*mut libsqlite3_sys::sqlite3, }; if err != libsqlite3_sys::SQLITE_OK { let message = sqlite_errmsg(connection, err); + sqlite_close(connection, true); return Err((err, message)); } @@ -227,6 +348,7 @@ fn sqlite_open_native(path: &SqlitePath) -> Result<*mut libsqlite3_sys::sqlite3, }; if err != libsqlite3_sys::SQLITE_OK { let message = sqlite_errmsg(connection, err); + sqlite_close(connection, true); return Err((err, message)); } @@ -245,20 +367,25 @@ fn sqlite_open_native(path: &SqlitePath) -> Result<*mut libsqlite3_sys::sqlite3, } } -fn sqlite_get_connection( - path: SqlitePath, -) -> Result<*mut libsqlite3_sys::sqlite3, (c_int, String)> { - SQLITE_CONNECTIONS.with(|cell| { - for (conn_path, connection) in cell.borrow().iter() { - if conn_path == &path { - return Ok(*connection); - } - } +fn sqlite_close(connection: *mut libsqlite3_sys::sqlite3, failed_open: bool) { + if connection.is_null() { + return; + } - let connection = sqlite_open_native(&path)?; - cell.borrow_mut().push((path, connection)); - Ok(connection) - }) + let result = unsafe { sqlite3_close_v2(connection) }; + debug_assert_eq!(result, libsqlite3_sys::SQLITE_OK); + + #[cfg(test)] + if failed_open && result == libsqlite3_sys::SQLITE_OK { + FAILED_OPEN_HANDLES_CLOSED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + + #[cfg(not(test))] + let _ = failed_open; +} + +fn sqlite_get_connection(path: SqlitePath) -> Result, (c_int, String)> { + SQLITE_CONNECTIONS.with(|cell| cell.borrow_mut().get(path)) } fn sqlite_value_integer(value: i64) -> SqliteValue { @@ -479,7 +606,7 @@ pub extern "C" fn hosted_sqlite_prepare( let mut stmt: *mut libsqlite3_sys::sqlite3_stmt = core::ptr::null_mut(); let err = unsafe { libsqlite3_sys::sqlite3_prepare_v2( - connection, + connection.as_ptr(), query_string.as_ptr() as *const c_char, query_string.len() as c_int, &mut stmt, @@ -487,7 +614,12 @@ pub extern "C" fn hosted_sqlite_prepare( ) }; if err != libsqlite3_sys::SQLITE_OK { - let message = sqlite_errmsg(connection, err); + let message = sqlite_errmsg(connection.as_ptr(), err); + if !stmt.is_null() { + unsafe { + libsqlite3_sys::sqlite3_finalize(stmt); + } + } return try_sqlite_prepare_err(sqlite_error(err, &message, roc_host)); } @@ -630,3 +762,186 @@ pub extern "C" fn hosted_sqlite_reset(handle: *mut u64) -> HostSqliteBindResult release_sqlite_stmt(handle, roc_host); result } + +#[cfg(test)] +mod tests { + use super::*; + use std::path::{Path, PathBuf}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + static NEXT_TEMP_DIR: AtomicUsize = AtomicUsize::new(0); + + fn clear_connection_cache() { + SQLITE_CONNECTIONS.with(|cell| cell.borrow_mut().entries.clear()); + } + + fn memory_path() -> SqlitePath { + #[cfg(unix)] + { + SqlitePath::Unix(b":memory:".to_vec()) + } + #[cfg(windows)] + { + SqlitePath::Windows(":memory:".encode_utf16().collect()) + } + #[cfg(not(any(unix, windows)))] + { + compile_error!("SQLite tests require a Unix or Windows host"); + } + } + + fn native_path(path: &Path) -> SqlitePath { + #[cfg(unix)] + { + use std::os::unix::ffi::OsStrExt; + SqlitePath::Unix(path.as_os_str().as_bytes().to_vec()) + } + #[cfg(windows)] + { + use std::os::windows::ffi::OsStrExt; + SqlitePath::Windows(path.as_os_str().encode_wide().collect()) + } + #[cfg(not(any(unix, windows)))] + { + let _ = path; + compile_error!("SQLite tests require a Unix or Windows host"); + } + } + + fn temp_dir(label: &str) -> PathBuf { + loop { + let sequence = NEXT_TEMP_DIR.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "basic-cli-sqlite-{label}-{}-{sequence}", + std::process::id() + )); + match std::fs::create_dir(&path) { + Ok(()) => return path, + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(error) => panic!("failed to create test directory: {error}"), + } + } + } + + fn prepare_statement(connection: Rc, query: &str) -> SqliteStatement { + let query = CString::new(query).unwrap(); + let mut stmt = core::ptr::null_mut(); + let result = unsafe { + libsqlite3_sys::sqlite3_prepare_v2( + connection.as_ptr(), + query.as_ptr(), + -1, + &mut stmt, + core::ptr::null_mut(), + ) + }; + assert_eq!(result, libsqlite3_sys::SQLITE_OK); + SqliteStatement { connection, stmt } + } + + fn exec(connection: &SqliteConnection, query: &str) { + let query = CString::new(query).unwrap(); + let result = unsafe { + libsqlite3_sys::sqlite3_exec( + connection.as_ptr(), + query.as_ptr(), + None, + core::ptr::null_mut(), + core::ptr::null_mut(), + ) + }; + assert_eq!(result, libsqlite3_sys::SQLITE_OK); + } + + #[test] + fn failed_open_handles_are_closed_immediately() { + clear_connection_cache(); + let directory = temp_dir("failed-open"); + let path = native_path(&directory.join("missing").join("database.sqlite")); + let before = FAILED_OPEN_HANDLES_CLOSED.load(Ordering::Relaxed); + + for _ in 0..8 { + assert!(sqlite_open_native(&path).is_err()); + } + + assert_eq!( + FAILED_OPEN_HANDLES_CLOSED.load(Ordering::Relaxed) - before, + 8 + ); + std::fs::remove_dir(directory).unwrap(); + } + + #[test] + fn unused_file_connections_are_bounded() { + clear_connection_cache(); + let directory = temp_dir("bounded-cache"); + let mut connections = Vec::new(); + + for index in 0..(MAX_CACHED_CONNECTIONS + 8) { + let path = native_path(&directory.join(format!("database-{index}.sqlite"))); + let connection = sqlite_get_connection(path).unwrap(); + connections.push(Rc::downgrade(&connection)); + } + + assert_eq!( + connections + .iter() + .filter(|connection| connection.strong_count() > 0) + .count(), + MAX_CACHED_CONNECTIONS + ); + + clear_connection_cache(); + assert!(connections + .iter() + .all(|connection| connection.strong_count() == 0)); + std::fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn live_statement_keeps_connection_open_after_cache_release() { + clear_connection_cache(); + let directory = temp_dir("live-statement"); + let path = native_path(&directory.join("database.sqlite")); + let connection = sqlite_get_connection(path).unwrap(); + let weak_connection = Rc::downgrade(&connection); + let statement = prepare_statement(Rc::clone(&connection), "SELECT 1"); + + clear_connection_cache(); + drop(connection); + assert!(weak_connection.strong_count() > 0); + + drop(statement); + assert_eq!(weak_connection.strong_count(), 0); + std::fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn memory_path_reuses_one_connection_and_database() { + clear_connection_cache(); + let first = sqlite_get_connection(memory_path()).unwrap(); + exec( + &first, + "CREATE TABLE values_table (value INTEGER); \ + INSERT INTO values_table VALUES (42);", + ); + let first_raw = first.as_ptr(); + drop(first); + + let second = sqlite_get_connection(memory_path()).unwrap(); + assert_eq!(second.as_ptr(), first_raw); + let statement = prepare_statement(Rc::clone(&second), "SELECT value FROM values_table"); + assert_eq!( + unsafe { libsqlite3_sys::sqlite3_step(statement.stmt) }, + libsqlite3_sys::SQLITE_ROW + ); + assert_eq!( + unsafe { libsqlite3_sys::sqlite3_column_int64(statement.stmt, 0) }, + 42 + ); + + drop(statement); + drop(second); + clear_connection_cache(); + } +}