use crate::channel::mpsc;
use crate::cx::Cx;
use crate::database::transaction::trace_database_transaction;
use crate::obligation::graded::{ObligationToken, TransactionKind};
use crate::runtime::blocking_pool::{BlockingPool, BlockingPoolHandle};
use crate::time::{sleep, wall_now};
use crate::types::{CancelReason, Outcome};
use parking_lot::Mutex;
use std::collections::BTreeMap;
use std::fmt;
use std::future::poll_fn;
use std::marker::PhantomData;
use std::path::{Component, Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock};
use std::task::Poll;
use std::time::Duration;
static SQLITE_POOL: OnceLock<BlockingPool> = OnceLock::new();
const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_millis(250);
const DEFAULT_STATEMENT_CACHE_CAPACITY: usize = 64;
const SQLITE_ROW_STREAM_CHANNEL_CAPACITY: usize = 1;
const SQLITE_ROW_STREAM_FULL_BACKOFF: Duration = Duration::from_millis(1);
fn sqlite_cancelled_reason(cx: &Cx) -> CancelReason {
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled"))
}
fn sqlite_error_is_interrupt(err: &SqliteError) -> bool {
match err {
SqliteError::Sqlite(msg) => {
let msg = msg.to_ascii_lowercase();
msg.contains("interrupt")
}
_ => false,
}
}
async fn sqlite_wait_retry_delay(cx: &Cx, delay: Duration) -> Result<(), CancelReason> {
if delay.is_zero() {
cx.checkpoint().map_err(|_| sqlite_cancelled_reason(cx))?;
crate::runtime::yield_now().await;
return cx.checkpoint().map_err(|_| sqlite_cancelled_reason(cx));
}
let now = cx
.timer_driver()
.map_or_else(wall_now, |driver| driver.now());
let mut sleeper = sleep(now, delay);
poll_fn(|task_cx| {
if cx.checkpoint().is_err() {
return Poll::Ready(Err(sqlite_cancelled_reason(cx)));
}
Pin::new(&mut sleeper).poll(task_cx).map(Ok)
})
.await
}
fn wal_checkpoint_i64(row: &SqliteRow, column: &str) -> Result<i64, SqliteError> {
row.get_i64(column).map_err(|err| {
SqliteError::WalCheckpointFailed(format!(
"WAL checkpoint status column {column:?} was missing or non-integer: {err}"
))
})
}
fn get_sqlite_pool() -> BlockingPoolHandle {
SQLITE_POOL.get_or_init(|| BlockingPool::new(1, 4)).handle()
}
fn configure_connection_defaults(
conn: &rusqlite::Connection,
enable_wal: bool,
) -> Result<(), SqliteError> {
configure_connection_defaults_with(conn, enable_wal, |_, error| {
SqliteError::Sqlite(error.to_string())
})
}
fn configure_connection_defaults_with<E, F>(
conn: &rusqlite::Connection,
enable_wal: bool,
mut map_error: F,
) -> Result<(), E>
where
F: FnMut(SqliteOperation, rusqlite::Error) -> E,
{
conn.busy_timeout(DEFAULT_BUSY_TIMEOUT)
.map_err(|error| map_error(SqliteOperation::Configure, error))?;
conn.pragma_update(None, "foreign_keys", "ON")
.map_err(|error| map_error(SqliteOperation::Configure, error))?;
if enable_wal {
conn.pragma_update(None, "journal_mode", "WAL")
.map_err(|error| map_error(SqliteOperation::Configure, error))?;
}
conn.set_prepared_statement_cache_capacity(DEFAULT_STATEMENT_CACHE_CAPACITY);
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TransactionState {
Autocommit,
InTransaction,
NeedsRollback,
RollingBack, }
#[derive(Debug, Default)]
struct BeginLifecycle {
abandoned: bool,
opened: bool,
generation: Option<u64>,
}
#[derive(Clone)]
struct BeginAttempt {
lifecycle: Arc<Mutex<BeginLifecycle>>,
transaction_state: Arc<Mutex<TransactionState>>,
transaction_generation: Arc<AtomicU64>,
}
impl BeginAttempt {
fn new(
transaction_state: Arc<Mutex<TransactionState>>,
transaction_generation: Arc<AtomicU64>,
) -> Self {
Self {
lifecycle: Arc::new(Mutex::new(BeginLifecycle::default())),
transaction_state,
transaction_generation,
}
}
fn abandon(&self) {
let mut lifecycle = self.lifecycle.lock();
lifecycle.abandoned = true;
let mut state = self.transaction_state.lock();
let owns_current_generation = lifecycle.generation.is_some_and(|generation| {
self.transaction_generation.load(Ordering::Acquire) == generation
});
if owns_current_generation || (!lifecycle.opened && *state == TransactionState::Autocommit)
{
*state = TransactionState::NeedsRollback;
}
}
fn finish_worker(
&self,
conn: &rusqlite::Connection,
result: Result<u64, SqliteError>,
) -> Result<u64, SqliteError> {
if result.is_ok() {
let mut lifecycle = self.lifecycle.lock();
lifecycle.opened = true;
let Some(generation) =
advance_transaction_generation(self.transaction_generation.as_ref())
else {
drop(lifecycle);
rollback_abandoned_begin_mutex_guarded(
conn,
self.transaction_state.as_ref(),
self.transaction_generation.as_ref(),
)?;
return Err(SqliteError::Sqlite(
"managed SQLite transaction generation exhausted".to_string(),
));
};
lifecycle.generation = Some(generation);
if lifecycle.abandoned {
drop(lifecycle);
rollback_abandoned_begin_mutex_guarded(
conn,
self.transaction_state.as_ref(),
self.transaction_generation.as_ref(),
)?;
} else {
*self.transaction_state.lock() = TransactionState::InTransaction;
}
}
result
}
fn finish_worker_diagnosed(
&self,
conn: &rusqlite::Connection,
operation: SqliteOperation,
result: Result<u64, SqliteOperationError>,
) -> Result<u64, SqliteOperationError> {
if result.is_ok() {
let mut lifecycle = self.lifecycle.lock();
lifecycle.opened = true;
let Some(generation) =
advance_transaction_generation(self.transaction_generation.as_ref())
else {
drop(lifecycle);
rollback_abandoned_begin_mutex_guarded(
conn,
self.transaction_state.as_ref(),
self.transaction_generation.as_ref(),
)
.map_err(|error| SqliteOperationError::from_legacy(operation, error))?;
return Err(SqliteOperationError::from_legacy(
operation,
SqliteError::Sqlite(
"managed SQLite transaction generation exhausted".to_string(),
),
));
};
lifecycle.generation = Some(generation);
if lifecycle.abandoned {
drop(lifecycle);
rollback_abandoned_begin_mutex_guarded(
conn,
self.transaction_state.as_ref(),
self.transaction_generation.as_ref(),
)
.map_err(|error| SqliteOperationError::from_legacy(operation, error))?;
} else {
*self.transaction_state.lock() = TransactionState::InTransaction;
}
}
result
}
}
enum TransactionWorkerEffect {
Begin(BeginAttempt),
Finish(TransactionFinishEffect),
}
impl TransactionWorkerEffect {
fn execute_worker(self, conn: &rusqlite::Connection, sql: &str) -> Result<u64, SqliteError> {
match self {
Self::Begin(attempt) => {
if attempt.transaction_generation.load(Ordering::Acquire) >= u64::MAX - 1 {
return Err(SqliteError::Sqlite(
"managed SQLite transaction generation exhausted".to_string(),
));
}
let result = conn
.execute(sql, [])
.map(|rows| rows as u64)
.map_err(|error| SqliteError::Sqlite(error.to_string()));
attempt.finish_worker(conn, result)
}
Self::Finish(mut effect) => effect.execute_worker(conn, sql),
}
}
fn execute_worker_diagnosed(
self,
conn: &rusqlite::Connection,
sql: &str,
operation: SqliteOperation,
) -> Result<u64, SqliteOperationError> {
match self {
Self::Begin(attempt) => {
if attempt.transaction_generation.load(Ordering::Acquire) >= u64::MAX - 1 {
return Err(SqliteOperationError::from_legacy(
operation,
SqliteError::Sqlite(
"managed SQLite transaction generation exhausted".to_string(),
),
));
}
let result = conn
.execute(sql, [])
.map(|rows| rows as u64)
.map_err(|error| SqliteOperationError::from_rusqlite(operation, error));
attempt.finish_worker_diagnosed(conn, operation, result)
}
Self::Finish(mut effect) => effect.execute_worker_diagnosed(conn, sql, operation),
}
}
}
#[derive(Clone, Copy)]
enum TransactionFinishKind {
Commit,
Rollback,
}
struct TransactionFinishEffect {
transaction_state: Arc<Mutex<TransactionState>>,
transaction_generation: Arc<AtomicU64>,
expected_generation: u64,
kind: TransactionFinishKind,
obligation: Option<ObligationToken<TransactionKind>>,
}
impl TransactionFinishEffect {
fn new(
transaction_state: Arc<Mutex<TransactionState>>,
transaction_generation: Arc<AtomicU64>,
expected_generation: u64,
kind: TransactionFinishKind,
obligation: Option<ObligationToken<TransactionKind>>,
) -> Self {
Self {
transaction_state,
transaction_generation,
expected_generation,
kind,
obligation,
}
}
fn execute_worker(
&mut self,
conn: &rusqlite::Connection,
sql: &str,
) -> Result<u64, SqliteError> {
if self.transaction_generation.load(Ordering::Acquire) != self.expected_generation {
if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
return Err(SqliteError::TransactionFinished);
}
if conn.is_autocommit() {
let _ = advance_transaction_generation(self.transaction_generation.as_ref());
*self.transaction_state.lock() = TransactionState::Autocommit;
if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
return Err(SqliteError::TransactionFinished);
}
let result = conn
.execute(sql, [])
.map(|rows| rows as u64)
.map_err(|error| SqliteError::Sqlite(error.to_string()));
if result.is_ok() {
let _ = advance_transaction_generation(self.transaction_generation.as_ref());
*self.transaction_state.lock() = TransactionState::Autocommit;
if let Some(token) = self.obligation.take() {
match self.kind {
TransactionFinishKind::Commit => {
let _ = token.commit();
}
TransactionFinishKind::Rollback => {
let _ = token.abort();
}
}
}
} else if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
result
}
fn execute_worker_diagnosed(
&mut self,
conn: &rusqlite::Connection,
sql: &str,
operation: SqliteOperation,
) -> Result<u64, SqliteOperationError> {
if self.transaction_generation.load(Ordering::Acquire) != self.expected_generation {
if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
return Err(SqliteOperationError::from_legacy(
operation,
SqliteError::TransactionFinished,
));
}
if conn.is_autocommit() {
let _ = advance_transaction_generation(self.transaction_generation.as_ref());
*self.transaction_state.lock() = TransactionState::Autocommit;
if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
return Err(SqliteOperationError::from_legacy(
operation,
SqliteError::TransactionFinished,
));
}
let result = conn
.execute(sql, [])
.map(|rows| rows as u64)
.map_err(|error| SqliteOperationError::from_rusqlite(operation, error));
if result.is_ok() {
let _ = advance_transaction_generation(self.transaction_generation.as_ref());
*self.transaction_state.lock() = TransactionState::Autocommit;
if let Some(token) = self.obligation.take() {
match self.kind {
TransactionFinishKind::Commit => {
let _ = token.commit();
}
TransactionFinishKind::Rollback => {
let _ = token.abort();
}
}
}
} else if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
result
}
}
impl Drop for TransactionFinishEffect {
fn drop(&mut self) {
if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
}
}
struct BeginDropGuard {
attempt: BeginAttempt,
armed: bool,
}
impl BeginDropGuard {
fn new(
transaction_state: Arc<Mutex<TransactionState>>,
transaction_generation: Arc<AtomicU64>,
) -> Self {
Self {
attempt: BeginAttempt::new(transaction_state, transaction_generation),
armed: true,
}
}
fn attempt(&self) -> BeginAttempt {
self.attempt.clone()
}
fn disarm(&mut self) {
self.armed = false;
}
fn opened_generation(&self) -> Option<u64> {
self.attempt.lifecycle.lock().generation
}
fn abandon(&mut self) {
if std::mem::replace(&mut self.armed, false) {
self.attempt.abandon();
}
}
}
impl Drop for BeginDropGuard {
fn drop(&mut self) {
self.abandon();
}
}
fn rollback_abandoned_begin_mutex_guarded(
conn: &rusqlite::Connection,
transaction_state: &Mutex<TransactionState>,
transaction_generation: &AtomicU64,
) -> Result<(), SqliteError> {
let mut state = transaction_state.lock();
*state = TransactionState::RollingBack;
if conn.is_autocommit() {
let _ = advance_transaction_generation(transaction_generation);
*state = TransactionState::Autocommit;
return Ok(());
}
match conn.execute_batch("ROLLBACK") {
Ok(()) => {
let _ = advance_transaction_generation(transaction_generation);
*state = TransactionState::Autocommit;
Ok(())
}
Err(_) if conn.is_autocommit() => {
let _ = advance_transaction_generation(transaction_generation);
*state = TransactionState::Autocommit;
Ok(())
}
Err(error) => {
*state = TransactionState::NeedsRollback;
Err(SqliteError::Sqlite(error.to_string()))
}
}
}
fn advance_transaction_generation(transaction_generation: &AtomicU64) -> Option<u64> {
transaction_generation
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |generation| {
generation.checked_add(1)
})
.ok()
.and_then(|generation| generation.checked_add(1))
}
fn rollback_orphaned_transaction_generation_guarded(
conn: &rusqlite::Connection,
transaction_state: &Mutex<TransactionState>,
transaction_generation: &AtomicU64,
) -> Result<(), SqliteError> {
if *transaction_state.lock() != TransactionState::NeedsRollback {
return Ok(());
}
rollback_orphaned_transaction_mutex_guarded(conn, transaction_state)?;
let _ = advance_transaction_generation(transaction_generation);
*transaction_state.lock() = TransactionState::Autocommit;
Ok(())
}
fn rollback_orphaned_transaction_mutex_guarded(
conn: &rusqlite::Connection,
transaction_state: &Mutex<TransactionState>,
) -> Result<(), SqliteError> {
let mut state_guard = transaction_state.lock();
if *state_guard != TransactionState::NeedsRollback {
return Ok(());
}
*state_guard = TransactionState::RollingBack;
drop(state_guard);
let final_state = if conn.is_autocommit() {
TransactionState::Autocommit
} else {
match conn.execute_batch("ROLLBACK") {
Ok(()) => TransactionState::Autocommit,
Err(e) => {
if conn.is_autocommit() {
TransactionState::Autocommit
} else {
let mut state_guard = transaction_state.lock();
*state_guard = TransactionState::NeedsRollback;
return Err(SqliteError::Sqlite(e.to_string()));
}
}
}
};
let mut state_guard = transaction_state.lock();
*state_guard = final_state;
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SqlSurfaceViolation {
Pragma,
TransactionControl,
AttachDetach,
Vacuum,
ExtensionLoading,
ParserRejected,
StatementCount,
ResourceLimit,
}
impl SqlSurfaceViolation {
fn checked_surface_message(self) -> &'static str {
match self {
Self::Pragma => "PRAGMA statements require the explicit *_unchecked SQLite APIs",
Self::TransactionControl => {
"transaction or connection control statements require the explicit *_unchecked SQLite APIs"
}
Self::AttachDetach => "ATTACH and DETACH are disabled on the checked SQLite APIs",
Self::Vacuum => {
"VACUUM requires the explicit *_unchecked SQLite APIs because VACUUM INTO can write an arbitrary filesystem path"
}
Self::ExtensionLoading => {
"SQLite extension loading is disabled on the checked SQLite APIs"
}
Self::ParserRejected => {
"checked SQLite SQL must be accepted by the bounded policy parser; audited engine-specific SQL requires an explicit *_unchecked API"
}
Self::StatementCount => "this checked SQLite API requires exactly one SQL statement",
Self::ResourceLimit => "SQLite SQL exceeds the checked-surface parser resource limits",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CheckedSqlCardinality {
ExactlyOne,
Batch,
}
const MAX_CHECKED_SQL_BYTES: usize = 1024 * 1024;
const MAX_CHECKED_SQL_RECURSION: usize = 128;
#[cfg(test)]
fn classify_sql_surface_violation(sql: &str) -> Option<SqlSurfaceViolation> {
match parse_checked_sql(sql) {
Ok(statements) => check_parsed_statements(&statements),
Err(violation) => Some(violation),
}
}
fn contains_extension_loading_call(sql: &str) -> Result<bool, SqlSurfaceViolation> {
use sqlparser::dialect::SQLiteDialect;
use sqlparser::tokenizer::{Token, Tokenizer};
let dialect = SQLiteDialect {};
let tokens = Tokenizer::new(&dialect, sql)
.tokenize()
.map_err(|_| SqlSurfaceViolation::ParserRejected)?;
let mut significant = tokens
.iter()
.filter(|token| !matches!(token, Token::Whitespace(_)))
.peekable();
while let Some(token) = significant.next() {
let Token::Word(word) = token else {
continue;
};
if word.value.eq_ignore_ascii_case("load_extension")
&& significant
.peek()
.is_some_and(|next| matches!(next, Token::LParen))
{
return Ok(true);
}
}
Ok(false)
}
fn unchecked_sql_contains_attach_detach(sql: &str) -> bool {
use sqlparser::ast::Statement;
use sqlparser::dialect::SQLiteDialect;
use sqlparser::parser::Parser;
let dialect = SQLiteDialect {};
match Parser::new(&dialect)
.with_recursion_limit(MAX_CHECKED_SQL_RECURSION)
.try_with_sql(sql)
.and_then(|mut parser| parser.parse_statements())
{
Ok(statements) => statements.iter().any(|statement| {
matches!(
statement,
Statement::AttachDatabase { .. }
| Statement::AttachDuckDBDatabase { .. }
| Statement::DetachDuckDBDatabase { .. }
)
}),
Err(_) => remove_sql_comments(sql).split(';').any(|statement| {
let statement = statement.trim().to_ascii_uppercase();
starts_with_sql_keyword(&statement, "ATTACH")
|| starts_with_sql_keyword(&statement, "DETACH")
}),
}
}
fn check_parsed_statements(
statements: &[sqlparser::ast::Statement],
) -> Option<SqlSurfaceViolation> {
use sqlparser::ast::Statement;
for statement in statements {
match statement {
Statement::Pragma { .. } => {
return Some(SqlSurfaceViolation::Pragma);
}
Statement::Set(_) if is_pragma_statement(statement) => {
return Some(SqlSurfaceViolation::Pragma);
}
Statement::AttachDatabase { .. }
| Statement::AttachDuckDBDatabase { .. }
| Statement::DetachDuckDBDatabase { .. } => {
return Some(SqlSurfaceViolation::AttachDetach);
}
Statement::Vacuum(_) => {
return Some(SqlSurfaceViolation::Vacuum);
}
Statement::StartTransaction { .. }
| Statement::Commit { .. }
| Statement::Rollback { .. }
| Statement::Savepoint { .. }
| Statement::ReleaseSavepoint { .. } => {
return Some(SqlSurfaceViolation::TransactionControl);
}
Statement::CreateTrigger { .. } => {
}
_ => {}
}
}
None
}
fn is_pragma_statement(statement: &sqlparser::ast::Statement) -> bool {
use sqlparser::ast::{ObjectName, Set, Statement};
fn name_is_pragma(name: &ObjectName) -> bool {
name.to_string().to_uppercase().starts_with("PRAGMA")
}
let Statement::Set(set) = statement else {
return false;
};
match set {
Set::SingleAssignment { variable, .. } => name_is_pragma(variable),
Set::ParenthesizedAssignments { variables, .. } => variables.iter().any(name_is_pragma),
Set::MultipleAssignments { assignments } => {
assignments.iter().any(|a| name_is_pragma(&a.name))
}
_ => false,
}
}
fn check_sql_keywords_fallback(sql: &str) -> Option<SqlSurfaceViolation> {
let sql_upper = sql.to_uppercase();
let sql_clean = remove_sql_comments(&sql_upper);
let statements: Vec<&str> = sql_clean.split(';').map(|s| s.trim()).collect();
for stmt in statements {
if stmt.is_empty() {
continue;
}
if starts_with_sql_keyword(stmt, "PRAGMA") {
return Some(SqlSurfaceViolation::Pragma);
}
if starts_with_sql_keyword(stmt, "ATTACH") || starts_with_sql_keyword(stmt, "DETACH") {
return Some(SqlSurfaceViolation::AttachDetach);
}
if starts_with_sql_keyword(stmt, "VACUUM") {
return Some(SqlSurfaceViolation::Vacuum);
}
if !stmt.contains(" TRIGGER ") {
if starts_with_sql_keyword(stmt, "BEGIN")
|| starts_with_sql_keyword(stmt, "COMMIT")
|| starts_with_sql_keyword(stmt, "ROLLBACK")
|| starts_with_sql_keyword(stmt, "SAVEPOINT")
|| starts_with_sql_keyword(stmt, "RELEASE")
|| starts_with_sql_keyword(stmt, "END")
{
return Some(SqlSurfaceViolation::TransactionControl);
}
}
}
None
}
fn starts_with_sql_keyword(statement: &str, keyword: &str) -> bool {
statement.strip_prefix(keyword).is_some_and(|suffix| {
suffix
.chars()
.next()
.is_none_or(|ch| !(ch.is_ascii_alphanumeric() || matches!(ch, '_' | '$')))
})
}
fn remove_sql_comments(sql: &str) -> String {
let mut result = String::with_capacity(sql.len());
let mut chars = sql.chars().peekable();
while let Some(ch) = chars.next() {
match ch {
'-' if chars.peek() == Some(&'-') => {
chars.next(); for ch in chars.by_ref() {
if ch == '\n' || ch == '\r' {
result.push(' ');
break;
}
}
}
'/' if chars.peek() == Some(&'*') => {
chars.next(); while let Some(ch) = chars.next() {
if ch == '*' && chars.peek() == Some(&'/') {
chars.next(); break;
}
}
result.push(' ');
}
'\'' | '"' | '`' => {
let quote = ch;
result.push(ch);
while let Some(ch) = chars.next() {
result.push(ch);
if ch == quote {
if chars.peek() == Some("e) {
chars.next(); result.push(quote);
} else {
break;
}
}
}
}
_ => result.push(ch),
}
}
result
}
fn parse_checked_sql(sql: &str) -> Result<Vec<sqlparser::ast::Statement>, SqlSurfaceViolation> {
use sqlparser::dialect::SQLiteDialect;
use sqlparser::parser::Parser;
if sql.len() > MAX_CHECKED_SQL_BYTES {
return Err(SqlSurfaceViolation::ResourceLimit);
}
let dialect = SQLiteDialect {};
let statements = Parser::new(&dialect)
.with_recursion_limit(MAX_CHECKED_SQL_RECURSION)
.try_with_sql(sql)
.and_then(|mut parser| parser.parse_statements())
.map_err(|_| {
check_sql_keywords_fallback(sql).unwrap_or(SqlSurfaceViolation::ParserRejected)
})?;
if contains_extension_loading_call(sql)? {
return Err(SqlSurfaceViolation::ExtensionLoading);
}
Ok(statements)
}
fn ensure_checked_sql_surface(
sql: &str,
cardinality: CheckedSqlCardinality,
) -> Result<(), SqliteError> {
let statements = parse_checked_sql(sql).map_err(|violation| {
SqliteError::UnsafeSql(violation.checked_surface_message().to_string())
})?;
if cardinality == CheckedSqlCardinality::ExactlyOne && statements.len() != 1 {
return Err(SqliteError::UnsafeSql(
SqlSurfaceViolation::StatementCount
.checked_surface_message()
.to_string(),
));
}
if let Some(violation) = check_parsed_statements(&statements) {
return Err(SqliteError::UnsafeSql(
violation.checked_surface_message().to_string(),
));
}
Ok(())
}
pub fn validate_checked_sql_statement(sql: &str) -> Result<(), SqliteError> {
ensure_checked_sql_surface(sql, CheckedSqlCardinality::ExactlyOne)
}
pub fn validate_checked_sql_batch(sql: &str) -> Result<(), SqliteError> {
ensure_checked_sql_surface(sql, CheckedSqlCardinality::Batch)
}
fn ensure_unchecked_sql_surface(sql: &str) -> Result<(), SqliteError> {
if unchecked_sql_contains_attach_detach(sql) {
return Err(SqliteError::UnsafeSql(
"ATTACH and DETACH are disabled on SQLite connections; open a separate validated connection instead"
.to_string(),
));
}
Ok(())
}
fn resolve_sqlite_open_path(path: &Path) -> Result<PathBuf, SqliteError> {
if path.exists() {
return std::fs::canonicalize(path).map_err(SqliteError::Io);
}
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
let canonical_parent = std::fs::canonicalize(parent).map_err(SqliteError::Io)?;
let file_name = path.file_name().ok_or_else(|| {
SqliteError::UnsafePath("SQLite database path must resolve to a file name".to_string())
})?;
Ok(canonical_parent.join(file_name))
}
fn validate_sqlite_open_path_lexical(path: &Path) -> Result<(), SqliteError> {
let raw = path.as_os_str().to_string_lossy();
if raw.starts_with('~') {
return Err(SqliteError::UnsafePath(
"tilde-prefixed SQLite paths are rejected; pass an explicit validated path".to_string(),
));
}
if path
.components()
.any(|component| matches!(component, Component::ParentDir))
{
return Err(SqliteError::UnsafePath(
"parent-directory traversal in SQLite paths is rejected; pass a normalized validated path"
.to_string(),
));
}
Ok(())
}
#[cfg(any(test, feature = "test-internals"))]
fn validate_sqlite_open_path(path: &Path) -> Result<(), SqliteError> {
validate_sqlite_open_path_lexical(path)?;
let resolved = resolve_sqlite_open_path(path)?;
validate_resolved_sqlite_path(&resolved)
}
fn validate_resolved_sqlite_path(resolved_path: &Path) -> Result<(), SqliteError> {
if resolved_path.starts_with(Path::new("/etc")) {
return Err(SqliteError::UnsafePath(format!(
"SQLite database path resolves into restricted system directory: {}",
resolved_path.display()
)));
}
if resolved_path.starts_with(Path::new("/sys")) {
return Err(SqliteError::UnsafePath(format!(
"SQLite database path resolves into restricted /sys directory: {}",
resolved_path.display()
)));
}
if resolved_path.starts_with(Path::new("/proc")) {
return Err(SqliteError::UnsafePath(format!(
"SQLite database path resolves into restricted /proc directory: {}",
resolved_path.display()
)));
}
if resolved_path.starts_with(Path::new("/dev")) {
return Err(SqliteError::UnsafePath(format!(
"SQLite database path resolves into restricted /dev directory: {}",
resolved_path.display()
)));
}
Ok(())
}
#[cfg(feature = "test-internals")]
#[doc(hidden)]
pub fn fuzz_validate_sqlite_open_path(path: &Path) -> Result<(), SqliteError> {
validate_sqlite_open_path(path)
}
#[derive(Debug)]
pub enum SqliteError {
Sqlite(String),
Cancelled(CancelReason),
ConnectionClosed,
ColumnNotFound(String),
TypeMismatch {
column: String,
expected: &'static str,
actual: String,
},
Io(std::io::Error),
TransactionFinished,
LockPoisoned,
UnsafeSql(String),
UnsafePath(String),
InvalidTextEncoding {
column: String,
source: std::str::Utf8Error,
},
WalCheckpointFailed(String),
StatementTimeout {
limit: std::time::Duration,
},
}
impl SqliteError {
#[must_use]
pub fn is_busy(&self) -> bool {
match self {
Self::Sqlite(msg) => msg.contains("database is locked") || msg.contains("SQLITE_BUSY"),
_ => false,
}
}
#[must_use]
pub fn is_locked(&self) -> bool {
match self {
Self::Sqlite(msg) => {
msg.contains("database table is locked") || msg.contains("SQLITE_LOCKED")
}
_ => false,
}
}
#[must_use]
pub fn is_constraint_violation(&self) -> bool {
match self {
Self::Sqlite(msg) => {
msg.contains("SQLITE_CONSTRAINT")
|| msg.contains("UNIQUE constraint failed")
|| msg.contains("NOT NULL constraint failed")
|| msg.contains("FOREIGN KEY constraint failed")
|| msg.contains("CHECK constraint failed")
}
_ => false,
}
}
#[must_use]
pub fn is_unique_violation(&self) -> bool {
match self {
Self::Sqlite(msg) => msg.contains("UNIQUE constraint failed"),
_ => false,
}
}
#[must_use]
pub fn is_connection_error(&self) -> bool {
matches!(
self,
Self::Io(_) | Self::ConnectionClosed | Self::LockPoisoned
)
}
#[must_use]
pub fn is_transient(&self) -> bool {
if matches!(self, Self::Io(_) | Self::ConnectionClosed) {
return true;
}
self.is_busy() || self.is_locked()
}
#[must_use]
pub fn is_retryable(&self) -> bool {
self.is_transient()
}
#[must_use]
pub fn error_code(&self) -> Option<&str> {
match self {
Self::Sqlite(msg) => {
if msg.contains("SQLITE_BUSY") || msg.contains("database is locked") {
Some("SQLITE_BUSY")
} else if msg.contains("SQLITE_LOCKED") || msg.contains("database table is locked")
{
Some("SQLITE_LOCKED")
} else if msg.contains("SQLITE_CONSTRAINT") || msg.contains("constraint failed") {
Some("SQLITE_CONSTRAINT")
} else if msg.contains("SQLITE_ERROR") {
Some("SQLITE_ERROR")
} else {
None
}
}
Self::Io(_) => Some("SQLITE_IOERR"),
Self::ConnectionClosed => Some("SQLITE_MISUSE"),
Self::UnsafePath(_) => Some("SQLITE_PERM"),
_ => None,
}
}
}
impl fmt::Display for SqliteError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Sqlite(msg) => write!(f, "SQLite error: {msg}"),
Self::Cancelled(reason) => write!(f, "SQLite operation cancelled: {reason:?}"),
Self::ConnectionClosed => write!(f, "SQLite connection is closed"),
Self::ColumnNotFound(name) => write!(f, "Column not found: {name}"),
Self::TypeMismatch {
column,
expected,
actual,
} => write!(
f,
"Type mismatch for column {column}: expected {expected}, got {actual}"
),
Self::Io(e) => write!(f, "SQLite I/O error: {e}"),
Self::TransactionFinished => write!(f, "Transaction already finished"),
Self::LockPoisoned => write!(f, "SQLite connection lock poisoned"),
Self::UnsafeSql(msg) => {
write!(
f,
"Unsafe SQLite control SQL on SQLite binding surface: {msg}"
)
}
Self::UnsafePath(msg) => write!(f, "Unsafe SQLite database path: {msg}"),
Self::InvalidTextEncoding { column, source } => {
write!(
f,
"SQLite text column {column} contained invalid UTF-8: {source}"
)
}
Self::WalCheckpointFailed(msg) => write!(f, "WAL checkpoint failed: {msg}"),
Self::StatementTimeout { limit } => write!(
f,
"statement aborted by budget-derived statement timeout ({limit:?})"
),
}
}
}
impl std::error::Error for SqliteError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(e) => Some(e),
Self::InvalidTextEncoding { source, .. } => Some(source),
_ => None,
}
}
}
impl From<std::io::Error> for SqliteError {
fn from(err: std::io::Error) -> Self {
Self::Io(err)
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SqliteOperation {
Open,
Prepare,
Bind,
Step,
ExecuteBatch,
TransactionBegin,
TransactionCommit,
TransactionRollback,
Configure,
Close,
BlockingPool,
Validation,
}
impl SqliteOperation {
#[must_use]
const fn as_str(self) -> &'static str {
match self {
Self::Open => "open",
Self::Prepare => "prepare",
Self::Bind => "bind",
Self::Step => "step",
Self::ExecuteBatch => "execute_batch",
Self::TransactionBegin => "transaction_begin",
Self::TransactionCommit => "transaction_commit",
Self::TransactionRollback => "transaction_rollback",
Self::Configure => "configure",
Self::Close => "close",
Self::BlockingPool => "blocking_pool",
Self::Validation => "validation",
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SqliteErrorCategory {
Busy,
Locked,
Constraint,
Interrupted,
Timeout,
PermissionDenied,
ReadOnly,
Io,
Corrupt,
ResourceExhausted,
InvalidInput,
NotFound,
Closed,
Cancelled,
Internal,
Unknown,
}
impl SqliteErrorCategory {
#[must_use]
const fn operator_code(self) -> &'static str {
match self {
Self::Busy => "sqlite.busy",
Self::Locked => "sqlite.locked",
Self::Constraint => "sqlite.constraint",
Self::Interrupted => "sqlite.interrupted",
Self::Timeout => "sqlite.timeout",
Self::PermissionDenied => "sqlite.permission_denied",
Self::ReadOnly => "sqlite.read_only",
Self::Io => "sqlite.io",
Self::Corrupt => "sqlite.corrupt",
Self::ResourceExhausted => "sqlite.resource_exhausted",
Self::InvalidInput => "sqlite.invalid_input",
Self::NotFound => "sqlite.not_found",
Self::Closed => "sqlite.closed",
Self::Cancelled => "sqlite.cancelled",
Self::Internal => "sqlite.internal",
Self::Unknown => "sqlite.unknown",
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SqliteRetryDisposition {
Never,
RetryOperation,
ReopenConnection,
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SqliteErrorDiagnostic {
operation: SqliteOperation,
category: SqliteErrorCategory,
primary_code: Option<&'static str>,
extended_code: Option<i32>,
retry: SqliteRetryDisposition,
connection_error: bool,
}
impl SqliteErrorDiagnostic {
#[must_use]
pub const fn operation(&self) -> SqliteOperation {
self.operation
}
#[must_use]
pub const fn category(&self) -> SqliteErrorCategory {
self.category
}
#[must_use]
pub const fn operator_code(&self) -> &'static str {
self.category.operator_code()
}
#[must_use]
pub const fn primary_code(&self) -> Option<&'static str> {
self.primary_code
}
#[must_use]
pub const fn extended_code(&self) -> Option<i32> {
self.extended_code
}
#[must_use]
pub const fn retry_disposition(&self) -> SqliteRetryDisposition {
self.retry
}
#[must_use]
pub const fn is_retryable(&self) -> bool {
matches!(self.retry, SqliteRetryDisposition::RetryOperation)
}
#[must_use]
pub const fn is_connection_error(&self) -> bool {
self.connection_error
}
fn from_legacy(operation: SqliteOperation, error: &SqliteError) -> Self {
let (category, retry, connection_error) = match error {
SqliteError::Cancelled(_) => (
SqliteErrorCategory::Cancelled,
SqliteRetryDisposition::Never,
false,
),
SqliteError::ConnectionClosed => (
SqliteErrorCategory::Closed,
SqliteRetryDisposition::ReopenConnection,
true,
),
SqliteError::ColumnNotFound(_) => (
SqliteErrorCategory::NotFound,
SqliteRetryDisposition::Never,
false,
),
SqliteError::TypeMismatch { .. }
| SqliteError::UnsafeSql(_)
| SqliteError::InvalidTextEncoding { .. }
| SqliteError::TransactionFinished => (
SqliteErrorCategory::InvalidInput,
SqliteRetryDisposition::Never,
false,
),
SqliteError::UnsafePath(_) => (
SqliteErrorCategory::PermissionDenied,
SqliteRetryDisposition::Never,
false,
),
SqliteError::Io(_) => (
SqliteErrorCategory::Io,
SqliteRetryDisposition::ReopenConnection,
true,
),
SqliteError::LockPoisoned => (
SqliteErrorCategory::Internal,
SqliteRetryDisposition::ReopenConnection,
true,
),
SqliteError::StatementTimeout { .. } => (
SqliteErrorCategory::Timeout,
SqliteRetryDisposition::Never,
false,
),
SqliteError::WalCheckpointFailed(_) => (
SqliteErrorCategory::Io,
SqliteRetryDisposition::RetryOperation,
false,
),
SqliteError::Sqlite(_) => (
SqliteErrorCategory::Unknown,
SqliteRetryDisposition::Never,
false,
),
};
Self {
operation,
category,
primary_code: None,
extended_code: None,
retry,
connection_error,
}
}
fn from_rusqlite(operation: SqliteOperation, error: &rusqlite::Error) -> Self {
let (code, extended_code) = match error {
rusqlite::Error::SqlInputError { error, .. } => {
(Some(error.code), Some(error.extended_code))
}
_ => (
error.sqlite_error_code(),
error.sqlite_extended_error_code(),
),
};
let (category, primary_code, retry, connection_error) = match code {
Some(rusqlite::ffi::ErrorCode::DatabaseBusy) => (
SqliteErrorCategory::Busy,
Some("SQLITE_BUSY"),
SqliteRetryDisposition::RetryOperation,
false,
),
Some(rusqlite::ffi::ErrorCode::DatabaseLocked) => (
SqliteErrorCategory::Locked,
Some("SQLITE_LOCKED"),
SqliteRetryDisposition::RetryOperation,
false,
),
Some(rusqlite::ffi::ErrorCode::ConstraintViolation) => (
SqliteErrorCategory::Constraint,
Some("SQLITE_CONSTRAINT"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::OperationInterrupted) => (
SqliteErrorCategory::Interrupted,
Some("SQLITE_INTERRUPT"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::PermissionDenied) => (
SqliteErrorCategory::PermissionDenied,
Some("SQLITE_PERM"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::AuthorizationForStatementDenied) => (
SqliteErrorCategory::PermissionDenied,
Some("SQLITE_AUTH"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::ReadOnly) => (
SqliteErrorCategory::ReadOnly,
Some("SQLITE_READONLY"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::SystemIoFailure) => (
SqliteErrorCategory::Io,
Some("SQLITE_IOERR"),
SqliteRetryDisposition::ReopenConnection,
true,
),
Some(rusqlite::ffi::ErrorCode::DatabaseCorrupt) => (
SqliteErrorCategory::Corrupt,
Some("SQLITE_CORRUPT"),
SqliteRetryDisposition::ReopenConnection,
true,
),
Some(rusqlite::ffi::ErrorCode::NotADatabase) => (
SqliteErrorCategory::Corrupt,
Some("SQLITE_NOTADB"),
SqliteRetryDisposition::ReopenConnection,
true,
),
Some(rusqlite::ffi::ErrorCode::OutOfMemory) => (
SqliteErrorCategory::ResourceExhausted,
Some("SQLITE_NOMEM"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::DiskFull) => (
SqliteErrorCategory::ResourceExhausted,
Some("SQLITE_FULL"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::TooBig) => (
SqliteErrorCategory::ResourceExhausted,
Some("SQLITE_TOOBIG"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::CannotOpen) => (
SqliteErrorCategory::Io,
Some("SQLITE_CANTOPEN"),
SqliteRetryDisposition::ReopenConnection,
true,
),
Some(rusqlite::ffi::ErrorCode::NotFound) => (
SqliteErrorCategory::NotFound,
Some("SQLITE_NOTFOUND"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::SchemaChanged) => (
SqliteErrorCategory::Internal,
Some("SQLITE_SCHEMA"),
SqliteRetryDisposition::RetryOperation,
false,
),
Some(rusqlite::ffi::ErrorCode::TypeMismatch) => (
SqliteErrorCategory::InvalidInput,
Some("SQLITE_MISMATCH"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::ParameterOutOfRange) => (
SqliteErrorCategory::InvalidInput,
Some("SQLITE_RANGE"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::ApiMisuse) => (
SqliteErrorCategory::Internal,
Some("SQLITE_MISUSE"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::OperationAborted) => (
SqliteErrorCategory::Internal,
Some("SQLITE_ABORT"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::FileLockingProtocolFailed) => (
SqliteErrorCategory::Io,
Some("SQLITE_PROTOCOL"),
SqliteRetryDisposition::ReopenConnection,
true,
),
Some(rusqlite::ffi::ErrorCode::InternalMalfunction) => (
SqliteErrorCategory::Internal,
Some("SQLITE_INTERNAL"),
SqliteRetryDisposition::ReopenConnection,
true,
),
Some(rusqlite::ffi::ErrorCode::NoLargeFileSupport) => (
SqliteErrorCategory::Internal,
Some("SQLITE_NOLFS"),
SqliteRetryDisposition::Never,
false,
),
Some(rusqlite::ffi::ErrorCode::Unknown) => (
if matches!(
operation,
SqliteOperation::Prepare
| SqliteOperation::Bind
| SqliteOperation::TransactionBegin
) {
SqliteErrorCategory::InvalidInput
} else {
SqliteErrorCategory::Unknown
},
Some("SQLITE_ERROR"),
SqliteRetryDisposition::Never,
false,
),
None => {
let operation = match error {
rusqlite::Error::InvalidParameterCount(_, _)
| rusqlite::Error::InvalidParameterName(_)
| rusqlite::Error::NulError(_)
| rusqlite::Error::ToSqlConversionFailure(_) => SqliteOperation::Bind,
_ => operation,
};
return Self {
operation,
category: SqliteErrorCategory::InvalidInput,
primary_code: None,
extended_code: None,
retry: SqliteRetryDisposition::Never,
connection_error: false,
};
}
Some(_) => (
SqliteErrorCategory::Unknown,
None,
SqliteRetryDisposition::Never,
false,
),
};
Self {
operation,
category,
primary_code,
extended_code,
retry,
connection_error,
}
}
}
pub struct SqliteOperationError {
diagnostic: SqliteErrorDiagnostic,
legacy: SqliteError,
engine_source: Option<rusqlite::Error>,
}
impl SqliteOperationError {
fn from_rusqlite(operation: SqliteOperation, error: rusqlite::Error) -> Self {
let diagnostic = SqliteErrorDiagnostic::from_rusqlite(operation, &error);
let rendered = error.to_string();
Self {
diagnostic,
legacy: SqliteError::Sqlite(rendered),
engine_source: Some(error),
}
}
fn from_legacy(operation: SqliteOperation, legacy: SqliteError) -> Self {
let diagnostic = SqliteErrorDiagnostic::from_legacy(operation, &legacy);
Self {
diagnostic,
legacy,
engine_source: None,
}
}
#[must_use]
pub const fn diagnostic(&self) -> &SqliteErrorDiagnostic {
&self.diagnostic
}
#[must_use]
pub const fn legacy_error(&self) -> &SqliteError {
&self.legacy
}
#[must_use]
pub fn engine_source(&self) -> Option<&(dyn std::error::Error + 'static)> {
self.engine_source
.as_ref()
.map(|error| error as &(dyn std::error::Error + 'static))
}
#[must_use]
pub fn into_legacy(self) -> SqliteError {
self.legacy
}
}
impl fmt::Debug for SqliteOperationError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("SqliteOperationError")
.field("diagnostic", &self.diagnostic)
.field("legacy", &"<redacted; call legacy_error() explicitly>")
.finish()
}
}
impl fmt::Display for SqliteOperationError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"[{}] SQLite {} failed",
self.diagnostic.operator_code(),
self.diagnostic.operation().as_str()
)?;
if let Some(primary) = self.diagnostic.primary_code() {
write!(f, " ({primary}")?;
if let Some(extended) = self.diagnostic.extended_code() {
write!(f, ", extended={extended}")?;
}
write!(f, ")")?;
}
Ok(())
}
}
impl std::error::Error for SqliteOperationError {}
fn diagnose_legacy_outcome<T>(
operation: SqliteOperation,
outcome: Outcome<T, SqliteError>,
) -> Outcome<T, SqliteOperationError> {
match outcome {
Outcome::Ok(value) => Outcome::Ok(value),
Outcome::Err(error) => Outcome::Err(SqliteOperationError::from_legacy(operation, error)),
Outcome::Cancelled(reason) => Outcome::Cancelled(reason),
Outcome::Panicked(payload) => Outcome::Panicked(payload),
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum SqliteValue {
Null,
Integer(i64),
Real(f64),
Text(String),
Blob(Vec<u8>),
}
impl SqliteValue {
#[must_use]
pub fn is_null(&self) -> bool {
matches!(self, Self::Null)
}
#[must_use]
pub fn as_integer(&self) -> Option<i64> {
match self {
Self::Integer(v) => Some(*v),
_ => None,
}
}
#[must_use]
pub fn as_real(&self) -> Option<f64> {
match self {
Self::Real(v) => Some(*v),
#[allow(clippy::cast_precision_loss)]
Self::Integer(v) => Some(*v as f64),
_ => None,
}
}
#[must_use]
pub fn as_real_strict(&self) -> Option<f64> {
match self {
Self::Real(v) => Some(*v),
_ => None,
}
}
#[must_use]
pub fn as_text(&self) -> Option<&str> {
match self {
Self::Text(v) => Some(v),
_ => None,
}
}
#[must_use]
pub fn as_blob(&self) -> Option<&[u8]> {
match self {
Self::Blob(v) => Some(v),
_ => None,
}
}
}
impl fmt::Display for SqliteValue {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Null => write!(f, "NULL"),
Self::Integer(v) => write!(f, "{v}"),
Self::Real(v) => write!(f, "{v}"),
Self::Text(v) => write!(f, "{v}"),
Self::Blob(v) => write!(f, "<blob {} bytes>", v.len()),
}
}
}
#[derive(Clone)]
pub struct SqliteRow {
columns: Arc<BTreeMap<String, usize>>,
ordered_columns: Arc<[String]>,
values: Vec<SqliteValue>,
}
impl fmt::Debug for SqliteRow {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("SqliteRow")
.field("columns", &self.columns)
.field("values", &self.values)
.finish()
}
}
impl SqliteRow {
fn new(
columns: Arc<BTreeMap<String, usize>>,
ordered_columns: Arc<[String]>,
values: Vec<SqliteValue>,
) -> Self {
Self {
columns,
ordered_columns,
values,
}
}
pub fn get(&self, column: &str) -> Result<&SqliteValue, SqliteError> {
let idx = self
.columns
.get(column)
.ok_or_else(|| SqliteError::ColumnNotFound(column.to_string()))?;
self.values
.get(*idx)
.ok_or_else(|| SqliteError::ColumnNotFound(column.to_string()))
}
pub fn get_idx(&self, idx: usize) -> Result<&SqliteValue, SqliteError> {
self.values
.get(idx)
.ok_or_else(|| SqliteError::ColumnNotFound(format!("index {idx}")))
}
pub fn get_i64(&self, column: &str) -> Result<i64, SqliteError> {
let val = self.get(column)?;
val.as_integer().ok_or_else(|| SqliteError::TypeMismatch {
column: column.to_string(),
expected: "integer",
actual: format!("{val:?}"),
})
}
pub fn get_f64(&self, column: &str) -> Result<f64, SqliteError> {
let val = self.get(column)?;
val.as_real().ok_or_else(|| SqliteError::TypeMismatch {
column: column.to_string(),
expected: "real",
actual: format!("{val:?}"),
})
}
pub fn get_f64_strict(&self, column: &str) -> Result<f64, SqliteError> {
let val = self.get(column)?;
val.as_real_strict()
.ok_or_else(|| SqliteError::TypeMismatch {
column: column.to_string(),
expected: "real",
actual: format!("{val:?}"),
})
}
pub fn get_str(&self, column: &str) -> Result<&str, SqliteError> {
let val = self.get(column)?;
val.as_text().ok_or_else(|| SqliteError::TypeMismatch {
column: column.to_string(),
expected: "text",
actual: format!("{val:?}"),
})
}
pub fn get_blob(&self, column: &str) -> Result<&[u8], SqliteError> {
let val = self.get(column)?;
val.as_blob().ok_or_else(|| SqliteError::TypeMismatch {
column: column.to_string(),
expected: "blob",
actual: format!("{val:?}"),
})
}
#[must_use]
pub fn len(&self) -> usize {
self.values.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.values.is_empty()
}
pub fn column_names(&self) -> impl Iterator<Item = &str> {
self.columns.keys().map(String::as_str)
}
pub fn column_names_in_order(&self) -> impl ExactSizeIterator<Item = &str> {
self.ordered_columns.iter().map(String::as_str)
}
#[must_use]
pub fn column_name(&self, index: usize) -> Option<&str> {
self.ordered_columns.get(index).map(String::as_str)
}
#[must_use]
pub fn column_index(&self, name: &str) -> Option<usize> {
self.ordered_columns
.iter()
.position(|column| column.eq_ignore_ascii_case(name))
}
}
#[derive(Debug, Default)]
struct SqliteRowStreamCounters {
rows_stepped: AtomicUsize,
rows_yielded: AtomicUsize,
buffered_rows: AtomicUsize,
peak_buffered_rows: AtomicUsize,
}
impl SqliteRowStreamCounters {
fn record_buffered_row(&self) {
let buffered = self
.buffered_rows
.fetch_add(1, Ordering::AcqRel)
.saturating_add(1);
let observed = buffered.min(SQLITE_ROW_STREAM_CHANNEL_CAPACITY);
let mut peak = self.peak_buffered_rows.load(Ordering::Acquire);
while observed > peak {
match self.peak_buffered_rows.compare_exchange_weak(
peak,
observed,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => break,
Err(current) => peak = current,
}
}
}
fn record_yielded_row(&self) {
self.buffered_rows.fetch_sub(1, Ordering::AcqRel);
self.rows_yielded.fetch_add(1, Ordering::AcqRel);
}
fn snapshot(&self) -> SqliteRowStreamStats {
SqliteRowStreamStats {
rows_stepped: self.rows_stepped.load(Ordering::Acquire),
rows_yielded: self.rows_yielded.load(Ordering::Acquire),
buffered_rows: self.buffered_rows.load(Ordering::Acquire),
peak_buffered_rows: self.peak_buffered_rows.load(Ordering::Acquire),
channel_capacity: SQLITE_ROW_STREAM_CHANNEL_CAPACITY,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SqliteRowStreamStats {
pub rows_stepped: usize,
pub rows_yielded: usize,
pub buffered_rows: usize,
pub peak_buffered_rows: usize,
pub channel_capacity: usize,
}
type SqliteRowStreamMessage = Result<SqliteRow, SqliteError>;
fn send_sqlite_stream_message(
sender: &mpsc::Sender<SqliteRowStreamMessage>,
counters: &SqliteRowStreamCounters,
mut message: SqliteRowStreamMessage,
) -> bool {
let is_row = message.is_ok();
loop {
match sender.try_reserve() {
Ok(permit) => {
if is_row {
counters.record_buffered_row();
}
match permit.send(message) {
Outcome::Ok(()) => return true,
Outcome::Err(
mpsc::SendError::Disconnected(_) | mpsc::SendError::Cancelled(_),
) => {
if is_row {
counters.buffered_rows.fetch_sub(1, Ordering::AcqRel);
}
return false;
}
Outcome::Err(mpsc::SendError::Full(value)) => {
if is_row {
counters.buffered_rows.fetch_sub(1, Ordering::AcqRel);
}
message = value;
}
Outcome::Cancelled(_) | Outcome::Panicked(_) => return false,
}
}
Err(mpsc::SendError::Disconnected(()) | mpsc::SendError::Cancelled(())) => {
return false;
}
Err(mpsc::SendError::Full(())) => {
std::thread::sleep(SQLITE_ROW_STREAM_FULL_BACKOFF);
}
}
}
}
fn sqlite_row_from_rusqlite_row(
row: &rusqlite::Row<'_>,
column_names: &Arc<[String]>,
columns: &Arc<BTreeMap<String, usize>>,
) -> Result<SqliteRow, SqliteError> {
let column_count = column_names.len();
let mut values = Vec::with_capacity(column_count);
for i in 0..column_count {
let value = row
.get_ref(i)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let column = column_name_or_index(column_names, i);
values.push(convert_value(value, &column)?);
}
Ok(SqliteRow::new(
Arc::clone(columns),
Arc::clone(column_names),
values,
))
}
fn sqlite_row_metadata(row: &rusqlite::Row<'_>) -> (Arc<[String]>, Arc<BTreeMap<String, usize>>) {
let column_names: Arc<[String]> = row
.as_ref()
.column_names()
.into_iter()
.map(str::to_owned)
.collect::<Vec<_>>()
.into();
let columns = column_names
.iter()
.enumerate()
.map(|(index, name)| (name.clone(), index))
.collect();
(column_names, Arc::new(columns))
}
pub struct SqliteRowStream<'connection> {
receiver: mpsc::Receiver<SqliteRowStreamMessage>,
handle: crate::runtime::blocking_pool::BlockingTaskHandle,
counters: Arc<SqliteRowStreamCounters>,
phase: Arc<Mutex<SqliteConnectionOpPhase>>,
finished: bool,
interrupt: Arc<rusqlite::InterruptHandle>,
_connection_lease: PhantomData<&'connection mut SqliteConnection>,
}
impl fmt::Debug for SqliteRowStream<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("SqliteRowStream")
.field("stats", &self.stats())
.field("finished", &self.finished)
.finish()
}
}
impl SqliteRowStream<'_> {
fn request_cancel(&self) -> SqliteConnectionOpPhase {
let mut phase = self.phase.lock();
let observed = *phase;
match observed {
SqliteConnectionOpPhase::Queued => {
*phase = SqliteConnectionOpPhase::CancelRequested;
}
SqliteConnectionOpPhase::Running => {
*phase = SqliteConnectionOpPhase::CancelRequested;
self.interrupt.interrupt();
}
SqliteConnectionOpPhase::CancelRequested | SqliteConnectionOpPhase::Completed => {}
}
observed
}
pub async fn next(&mut self, cx: &Cx) -> Outcome<Option<SqliteRow>, SqliteError> {
if self.finished {
return Outcome::Ok(None);
}
if cx.checkpoint().is_err() {
self.cancel_in_drain(cx).await;
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.receiver.recv(cx).await {
Ok(Ok(row)) => {
self.counters.record_yielded_row();
Outcome::Ok(Some(row))
}
Ok(Err(err)) => {
self.finish();
Outcome::Err(err)
}
Err(mpsc::RecvError::Disconnected) => {
self.finished = true;
Outcome::Ok(None)
}
Err(mpsc::RecvError::Cancelled) => {
self.cancel_in_drain(cx).await;
Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
)
}
Err(mpsc::RecvError::Empty) => Outcome::Err(SqliteError::Sqlite(
"sqlite row stream receive unexpectedly returned empty".to_string(),
)),
}
}
async fn cancel_in_drain(&mut self, cx: &Cx) {
const MASKED_DRAIN_POLLS: u32 = 1024;
if self.finished {
return;
}
self.finished = true;
let cancel_phase = self.request_cancel();
self.handle.cancel();
if !self.handle.is_done() {
match cancel_phase {
SqliteConnectionOpPhase::Running => cx.trace(
"client.wire_cancel proto=sqlite outcome=interrupt_sent op=row_stream",
),
SqliteConnectionOpPhase::Queued => cx.trace(
"client.wire_cancel proto=sqlite outcome=skipped op=row_stream reason=queued",
),
SqliteConnectionOpPhase::Completed => cx.trace(
"client.wire_cancel proto=sqlite outcome=skipped op=row_stream reason=completed",
),
SqliteConnectionOpPhase::CancelRequested => {}
}
let drained = crate::combinator::commit_section(cx, MASKED_DRAIN_POLLS, async {
loop {
match self.receiver.recv(cx).await {
Ok(_) => {
if self.handle.is_done() {
break true;
}
}
Err(mpsc::RecvError::Disconnected) => break true,
Err(_) => break false,
}
}
})
.await;
if drained {
cx.trace("client.wire_cancel proto=sqlite drain=job_completed op=row_stream");
} else {
cx.trace(
"client.wire_cancel proto=sqlite drain=masked_poll_budget_exhausted \
fallback=abandon_job op=row_stream",
);
}
}
self.receiver.close();
}
#[must_use]
pub fn stats(&self) -> SqliteRowStreamStats {
self.counters.snapshot()
}
fn finish(&mut self) {
if !self.finished {
self.finished = true;
self.receiver.close();
self.request_cancel();
self.handle.cancel();
}
}
}
impl Drop for SqliteRowStream<'_> {
fn drop(&mut self) {
self.finish();
}
}
struct SqliteConnectionInner {
conn: Option<rusqlite::Connection>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SqliteConnectionOpPhase {
Queued,
Running,
CancelRequested,
Completed,
}
enum SqliteConnectionOpCompletion<R, E> {
Finished(Result<R, E>),
Cancelled,
}
trait SqliteConnectionOpError: Send + 'static {
fn from_legacy(operation: SqliteOperation, error: SqliteError) -> Self;
fn is_interrupt(&self) -> bool;
fn statement_timeout(operation: SqliteOperation, limit: Duration) -> Self;
}
impl SqliteConnectionOpError for SqliteError {
fn from_legacy(_operation: SqliteOperation, error: SqliteError) -> Self {
error
}
fn is_interrupt(&self) -> bool {
sqlite_error_is_interrupt(self)
}
fn statement_timeout(_operation: SqliteOperation, limit: Duration) -> Self {
Self::StatementTimeout { limit }
}
}
impl SqliteConnectionOpError for SqliteOperationError {
fn from_legacy(operation: SqliteOperation, error: SqliteError) -> Self {
SqliteOperationError::from_legacy(operation, error)
}
fn is_interrupt(&self) -> bool {
self.diagnostic.category() == SqliteErrorCategory::Interrupted
}
fn statement_timeout(operation: SqliteOperation, limit: Duration) -> Self {
SqliteOperationError::from_legacy(operation, SqliteError::StatementTimeout { limit })
}
}
impl SqliteConnectionInner {
fn new(conn: rusqlite::Connection) -> Self {
Self { conn: Some(conn) }
}
fn get(&self) -> Result<&rusqlite::Connection, SqliteError> {
self.conn.as_ref().ok_or(SqliteError::ConnectionClosed)
}
fn close(&mut self) {
self.conn = None;
}
}
pub struct SqliteConnection {
inner: Arc<Mutex<SqliteConnectionInner>>,
pool: BlockingPoolHandle,
transaction_state: Arc<Mutex<TransactionState>>,
transaction_generation: Arc<AtomicU64>,
interrupt: Arc<rusqlite::InterruptHandle>,
statement_timeout_override: Option<std::time::Duration>,
}
impl fmt::Debug for SqliteConnection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let state = *self.transaction_state.lock();
f.debug_struct("SqliteConnection")
.field("open", &self.inner.lock().conn.is_some())
.field("pool", &self.pool)
.field("transaction_state", &state)
.field(
"transaction_generation",
&self.transaction_generation.load(Ordering::Acquire),
)
.finish()
}
}
impl SqliteConnection {
pub fn interrupt(&self) {
self.interrupt.interrupt();
}
pub fn set_statement_timeout_override(&mut self, timeout: Option<std::time::Duration>) {
self.statement_timeout_override = timeout;
}
#[must_use]
pub fn statement_timeout_override(&self) -> Option<std::time::Duration> {
self.statement_timeout_override
}
async fn run_connection_op<R, F>(
&self,
cx: &Cx,
op_name: &'static str,
f: F,
) -> Outcome<R, SqliteError>
where
R: Send + 'static,
F: FnOnce(&rusqlite::Connection) -> Result<R, SqliteError> + Send + 'static,
{
self.run_connection_op_inner(cx, op_name, SqliteOperation::BlockingPool, f)
.await
}
async fn run_connection_op_diagnosed<R, F>(
&self,
cx: &Cx,
op_name: &'static str,
operation: SqliteOperation,
f: F,
) -> Outcome<R, SqliteOperationError>
where
R: Send + 'static,
F: FnOnce(&rusqlite::Connection) -> Result<R, SqliteOperationError> + Send + 'static,
{
self.run_connection_op_inner(cx, op_name, operation, f)
.await
}
async fn run_connection_op_inner<R, E, F>(
&self,
cx: &Cx,
op_name: &'static str,
operation: SqliteOperation,
f: F,
) -> Outcome<R, E>
where
R: Send + 'static,
E: SqliteConnectionOpError,
F: FnOnce(&rusqlite::Connection) -> Result<R, E> + Send + 'static,
{
const TIMEOUT_PROGRESS_OPS: i32 = 1000;
const MASKED_DRAIN_POLLS: u32 = 1024;
let timeout =
crate::database::effective_statement_timeout(cx, self.statement_timeout_override);
if let Some(limit) = timeout {
let remaining_ns = crate::database::remaining_budget(cx)
.map_or_else(|| "none".to_string(), |d| d.as_nanos().to_string());
let base_ms = self.statement_timeout_override.map_or_else(
|| "none".to_string(),
|d| crate::database::statement_timeout_millis(d).to_string(),
);
cx.trace(&format!(
"client.budget_forwarded proto=sqlite base_ms={base_ms} \
remaining_ns={remaining_ns} statement_timeout_ms={}",
crate::database::statement_timeout_millis(limit)
));
}
let inner = Arc::clone(&self.inner);
let phase = Arc::new(Mutex::new(SqliteConnectionOpPhase::Queued));
let worker_phase = Arc::clone(&phase);
let (tx, mut rx) = crate::channel::oneshot::channel();
let permit = match tx.reserve(cx) {
Ok(permit) => permit,
Err(crate::channel::oneshot::SendError::Cancelled(())) => {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
Err(crate::channel::oneshot::SendError::Disconnected(())) => {
return Outcome::Err(E::from_legacy(
operation,
SqliteError::Sqlite(format!("failed to reserve result channel for {op_name}")),
));
}
};
let handle = self.pool.spawn(move || {
let completion = (|| {
let guard = inner.lock();
{
let mut phase = worker_phase.lock();
match *phase {
SqliteConnectionOpPhase::Queued => {
*phase = SqliteConnectionOpPhase::Running;
}
SqliteConnectionOpPhase::CancelRequested => {
*phase = SqliteConnectionOpPhase::Completed;
drop(phase);
drop(guard);
return SqliteConnectionOpCompletion::Cancelled;
}
SqliteConnectionOpPhase::Running | SqliteConnectionOpPhase::Completed => {
unreachable!("a SQLite connection operation starts exactly once")
}
}
}
let result = (|| {
let conn = guard
.get()
.map_err(|error| E::from_legacy(operation, error))?;
if let Some(limit) = timeout {
let deadline = std::time::Instant::now() + limit;
conn.progress_handler(
TIMEOUT_PROGRESS_OPS,
Some(move || std::time::Instant::now() >= deadline),
)
.map_err(|e| {
E::from_legacy(
operation,
SqliteError::Sqlite(format!(
"failed to arm statement timeout: {e}"
)),
)
})?;
}
let result = f(conn);
if timeout.is_some() {
let _ = conn.progress_handler(0, None::<fn() -> bool>);
}
result
})();
let cancellation_requested = {
let mut phase = worker_phase.lock();
let cancellation_requested = *phase == SqliteConnectionOpPhase::CancelRequested;
*phase = SqliteConnectionOpPhase::Completed;
cancellation_requested
};
let result = match (cancellation_requested, timeout, result) {
(true, _, Err(err)) if err.is_interrupt() => {
drop(guard);
return SqliteConnectionOpCompletion::Cancelled;
}
(_, Some(limit), Err(err)) if err.is_interrupt() => {
Err(E::statement_timeout(operation, limit))
}
(_, _, result) => result,
};
drop(guard);
SqliteConnectionOpCompletion::Finished(result)
})();
let _ = permit.send(completion);
});
match rx.recv(cx).await {
Ok(SqliteConnectionOpCompletion::Finished(Ok(result))) => Outcome::Ok(result),
Ok(SqliteConnectionOpCompletion::Finished(Err(e))) => Outcome::Err(e),
Ok(SqliteConnectionOpCompletion::Cancelled) => Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
),
Err(crate::channel::oneshot::RecvError::Cancelled) => {
let cancel_phase = {
let mut phase = phase.lock();
let observed = *phase;
match observed {
SqliteConnectionOpPhase::Queued => {
*phase = SqliteConnectionOpPhase::CancelRequested;
}
SqliteConnectionOpPhase::Running => {
*phase = SqliteConnectionOpPhase::CancelRequested;
self.interrupt.interrupt();
}
SqliteConnectionOpPhase::CancelRequested
| SqliteConnectionOpPhase::Completed => {}
}
observed
};
handle.cancel();
match cancel_phase {
SqliteConnectionOpPhase::Running => cx.trace(&format!(
"client.wire_cancel proto=sqlite outcome=interrupt_sent op={op_name}"
)),
SqliteConnectionOpPhase::Queued => cx.trace(&format!(
"client.wire_cancel proto=sqlite outcome=skipped op={op_name} reason=queued"
)),
SqliteConnectionOpPhase::Completed => cx.trace(&format!(
"client.wire_cancel proto=sqlite outcome=skipped op={op_name} reason=completed"
)),
SqliteConnectionOpPhase::CancelRequested => {}
}
let drained =
crate::combinator::commit_section(cx, MASKED_DRAIN_POLLS, rx.recv(cx)).await;
match drained {
Ok(SqliteConnectionOpCompletion::Finished(result)) => {
cx.trace(
"client.wire_cancel proto=sqlite drain=job_completed completion=won",
);
return match result {
Ok(result) => Outcome::Ok(result),
Err(err) => Outcome::Err(err),
};
}
Ok(SqliteConnectionOpCompletion::Cancelled) => {
cx.trace("client.wire_cancel proto=sqlite drain=job_cancelled");
}
Err(crate::channel::oneshot::RecvError::Closed) => {
cx.trace("client.wire_cancel proto=sqlite drain=job_not_started");
}
Err(_) => cx.trace(
"client.wire_cancel proto=sqlite drain=masked_poll_budget_exhausted \
fallback=abandon_job",
),
}
Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
)
}
Err(crate::channel::oneshot::RecvError::Closed) => Outcome::Err(E::from_legacy(
operation,
SqliteError::Sqlite(format!("failed to receive result for {op_name}")),
)),
Err(crate::channel::oneshot::RecvError::PolledAfterCompletion) => {
unreachable!("{op_name} awaits a fresh oneshot recv future")
}
}
}
async fn drain_orphaned_transaction(&self, cx: &Cx) -> Outcome<(), SqliteError> {
let current_state = *self.transaction_state.lock();
if current_state != TransactionState::NeedsRollback {
return Outcome::Ok(());
}
let transaction_state = Arc::clone(&self.transaction_state);
let transaction_generation = Arc::clone(&self.transaction_generation);
self.run_connection_op(cx, "sqlite rollback cleanup", move |conn| {
rollback_orphaned_transaction_generation_guarded(
conn,
transaction_state.as_ref(),
transaction_generation.as_ref(),
)
})
.await
}
fn schedule_dropped_transaction_rollback(&self, expected_generation: u64) {
let inner = Arc::clone(&self.inner);
let transaction_state = Arc::clone(&self.transaction_state);
let transaction_generation = Arc::clone(&self.transaction_generation);
let _cleanup = self.pool.spawn(move || {
let guard = inner.lock();
let Some(conn) = guard.conn.as_ref() else {
if transaction_generation.load(Ordering::Acquire) == expected_generation {
let _ = advance_transaction_generation(transaction_generation.as_ref());
*transaction_state.lock() = TransactionState::Autocommit;
}
return;
};
if transaction_generation.load(Ordering::Acquire) != expected_generation {
return;
}
let _ = rollback_orphaned_transaction_generation_guarded(
conn,
transaction_state.as_ref(),
transaction_generation.as_ref(),
);
});
}
async fn open_with<E, F>(cx: &Cx, operation: SqliteOperation, open: F) -> Outcome<Self, E>
where
E: SqliteConnectionOpError,
F: FnOnce() -> Result<rusqlite::Connection, E> + Send + 'static,
{
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
let pool = get_sqlite_pool();
let pool_clone = pool.clone();
let (tx, mut rx) = crate::channel::oneshot::channel();
let permit = tx.reserve(cx);
let handle = pool.spawn(move || {
let result = open();
if let Ok(permit) = permit {
let _ = permit.send(result);
}
});
match rx.recv(cx).await {
Ok(Ok(conn)) => {
let interrupt = Arc::new(conn.get_interrupt_handle());
Outcome::Ok(Self {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(conn))),
pool: pool_clone,
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
})
}
Ok(Err(error)) => Outcome::Err(error),
Err(crate::channel::oneshot::RecvError::Cancelled) => {
handle.cancel();
Outcome::Cancelled(sqlite_cancelled_reason(cx))
}
Err(crate::channel::oneshot::RecvError::Closed) => Outcome::Err(E::from_legacy(
operation,
SqliteError::Sqlite("failed to receive result".to_string()),
)),
Err(crate::channel::oneshot::RecvError::PolledAfterCompletion) => {
unreachable!("SQLite blocking-pool open awaits a fresh oneshot recv future")
}
}
}
pub async fn open(cx: &Cx, path: impl AsRef<Path>) -> Outcome<Self, SqliteError> {
let path = path.as_ref().to_path_buf();
Self::open_with(cx, SqliteOperation::Open, move || {
validate_sqlite_open_path_lexical(&path)?;
let resolved_path = resolve_sqlite_open_path(&path)?;
validate_resolved_sqlite_path(&resolved_path)?;
let conn = rusqlite::Connection::open(&resolved_path)
.map_err(|error| SqliteError::Sqlite(error.to_string()))?;
configure_connection_defaults(&conn, true)?;
Ok(conn)
})
.await
}
pub async fn open_diagnosed(
cx: &Cx,
path: impl AsRef<Path>,
) -> Outcome<Self, SqliteOperationError> {
let path = path.as_ref().to_path_buf();
Self::open_with(cx, SqliteOperation::Open, move || {
validate_sqlite_open_path_lexical(&path).map_err(|error| {
SqliteOperationError::from_legacy(SqliteOperation::Validation, error)
})?;
let resolved_path = resolve_sqlite_open_path(&path).map_err(|error| {
SqliteOperationError::from_legacy(SqliteOperation::Validation, error)
})?;
validate_resolved_sqlite_path(&resolved_path).map_err(|error| {
SqliteOperationError::from_legacy(SqliteOperation::Validation, error)
})?;
let conn = rusqlite::Connection::open(&resolved_path).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Open, error)
})?;
configure_connection_defaults_with(&conn, true, |operation, error| {
SqliteOperationError::from_rusqlite(operation, error)
})?;
Ok(conn)
})
.await
}
pub async fn open_in_memory(cx: &Cx) -> Outcome<Self, SqliteError> {
Self::open_with(cx, SqliteOperation::Open, move || {
let conn = rusqlite::Connection::open_in_memory()
.map_err(|error| SqliteError::Sqlite(error.to_string()))?;
configure_connection_defaults(&conn, false)?;
Ok(conn)
})
.await
}
pub async fn open_in_memory_diagnosed(cx: &Cx) -> Outcome<Self, SqliteOperationError> {
Self::open_with(cx, SqliteOperation::Open, move || {
let conn = rusqlite::Connection::open_in_memory().map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Open, error)
})?;
configure_connection_defaults_with(&conn, false, |operation, error| {
SqliteOperationError::from_rusqlite(operation, error)
})?;
Ok(conn)
})
.await
}
pub async fn execute(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteError> {
if let Err(err) = validate_checked_sql_statement(sql) {
return Outcome::Err(err);
}
self.execute_unchecked(cx, sql, params).await
}
pub async fn execute_unchecked(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteError> {
self.execute_unchecked_with(cx, sql, params, |conn, sql, params| {
let params_refs: Vec<&dyn rusqlite::ToSql> = params
.iter()
.map(|value| value as &dyn rusqlite::ToSql)
.collect();
conn.execute(sql, params_refs.as_slice())
.map(|rows| rows as u64)
.map_err(|error| SqliteError::Sqlite(error.to_string()))
})
.await
}
pub async fn execute_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteOperationError> {
if let Err(error) = validate_checked_sql_statement(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
self.execute_unchecked_diagnosed(cx, sql, params).await
}
pub async fn execute_unchecked_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteOperationError> {
if let Err(error) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
match diagnose_legacy_outcome(
SqliteOperation::TransactionRollback,
self.drain_orphaned_transaction(cx).await,
) {
Outcome::Ok(()) => {}
Outcome::Err(error) => return Outcome::Err(error),
Outcome::Cancelled(reason) => return Outcome::Cancelled(reason),
Outcome::Panicked(payload) => return Outcome::Panicked(payload),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
let sql = sql.to_string();
let params = params.to_vec();
self.run_connection_op_diagnosed(
cx,
"sqlite diagnosed execute",
SqliteOperation::Step,
move |conn| {
let mut statement = conn.prepare_cached(&sql).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Prepare, error)
})?;
let params_refs: Vec<&dyn rusqlite::ToSql> = params
.iter()
.map(|value| value as &dyn rusqlite::ToSql)
.collect();
statement
.execute(params_refs.as_slice())
.map(|rows| rows as u64)
.map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Step, error)
})
},
)
.await
}
async fn execute_transaction_control(
&self,
cx: &Cx,
sql: &'static str,
effect: TransactionWorkerEffect,
) -> Outcome<u64, SqliteError> {
self.execute_unchecked_with(cx, sql, &[], move |conn, sql, _params| {
effect.execute_worker(conn, sql)
})
.await
}
async fn execute_transaction_control_diagnosed(
&self,
cx: &Cx,
sql: &'static str,
operation: SqliteOperation,
effect: TransactionWorkerEffect,
) -> Outcome<u64, SqliteOperationError> {
if let Err(error) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
match diagnose_legacy_outcome(
SqliteOperation::TransactionRollback,
self.drain_orphaned_transaction(cx).await,
) {
Outcome::Ok(()) => {}
Outcome::Err(error) => return Outcome::Err(error),
Outcome::Cancelled(reason) => return Outcome::Cancelled(reason),
Outcome::Panicked(payload) => return Outcome::Panicked(payload),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
self.run_connection_op_diagnosed(
cx,
"sqlite diagnosed transaction control",
operation,
move |conn| effect.execute_worker_diagnosed(conn, sql, operation),
)
.await
}
async fn execute_unchecked_with<F>(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
execute: F,
) -> Outcome<u64, SqliteError>
where
F: FnOnce(&rusqlite::Connection, &str, &[SqliteValue]) -> Result<u64, SqliteError>
+ Send
+ 'static,
{
if let Err(err) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(err);
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.drain_orphaned_transaction(cx).await {
Outcome::Ok(()) => {}
Outcome::Err(e) => return Outcome::Err(e),
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
let sql = sql.to_string();
let params: Vec<SqliteValue> = params.to_vec();
self.run_connection_op(cx, "sqlite execute", move |conn| {
execute(conn, &sql, ¶ms)
})
.await
}
pub async fn execute_batch(&self, cx: &Cx, sql: &str) -> Outcome<(), SqliteError> {
if let Err(err) = validate_checked_sql_batch(sql) {
return Outcome::Err(err);
}
self.execute_batch_unchecked(cx, sql).await
}
pub async fn execute_batch_unchecked(&self, cx: &Cx, sql: &str) -> Outcome<(), SqliteError> {
if let Err(err) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(err);
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.drain_orphaned_transaction(cx).await {
Outcome::Ok(()) => {}
Outcome::Err(e) => return Outcome::Err(e),
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
let sql = sql.to_string();
self.run_connection_op(cx, "sqlite execute_batch", move |conn| {
conn.execute_batch(&sql)
.map_err(|e| SqliteError::Sqlite(e.to_string()))
})
.await
}
pub async fn execute_batch_diagnosed(
&self,
cx: &Cx,
sql: &str,
) -> Outcome<(), SqliteOperationError> {
if let Err(error) = validate_checked_sql_batch(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
self.execute_batch_unchecked_diagnosed(cx, sql).await
}
pub async fn execute_batch_unchecked_diagnosed(
&self,
cx: &Cx,
sql: &str,
) -> Outcome<(), SqliteOperationError> {
if let Err(error) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
match diagnose_legacy_outcome(
SqliteOperation::TransactionRollback,
self.drain_orphaned_transaction(cx).await,
) {
Outcome::Ok(()) => {}
Outcome::Err(error) => return Outcome::Err(error),
Outcome::Cancelled(reason) => return Outcome::Cancelled(reason),
Outcome::Panicked(payload) => return Outcome::Panicked(payload),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
let sql = sql.to_string();
self.run_connection_op_diagnosed(
cx,
"sqlite diagnosed execute_batch",
SqliteOperation::ExecuteBatch,
move |conn| {
conn.execute_batch(&sql).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::ExecuteBatch, error)
})
},
)
.await
}
pub async fn query(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Vec<SqliteRow>, SqliteError> {
if let Err(err) = validate_checked_sql_statement(sql) {
return Outcome::Err(err);
}
self.query_unchecked(cx, sql, params).await
}
pub async fn query_unchecked(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Vec<SqliteRow>, SqliteError> {
if let Err(err) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(err);
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.drain_orphaned_transaction(cx).await {
Outcome::Ok(()) => {}
Outcome::Err(e) => return Outcome::Err(e),
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
let sql = sql.to_string();
let params: Vec<SqliteValue> = params.to_vec();
self.run_connection_op(cx, "sqlite query", move |conn| {
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(|v| v as &dyn rusqlite::ToSql).collect();
let mut stmt = conn
.prepare_cached(&sql)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let mut rows = stmt
.query(params_refs.as_slice())
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let mut result = Vec::new();
let mut metadata = None;
while let Some(row) = rows
.next()
.map_err(|e| SqliteError::Sqlite(e.to_string()))?
{
let (column_names, columns) =
metadata.get_or_insert_with(|| sqlite_row_metadata(row));
result.push(sqlite_row_from_rusqlite_row(row, column_names, columns)?);
}
drop(rows);
drop(stmt);
Ok(result)
})
.await
}
pub async fn query_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Vec<SqliteRow>, SqliteOperationError> {
if let Err(error) = validate_checked_sql_statement(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
self.query_unchecked_diagnosed(cx, sql, params).await
}
pub async fn query_unchecked_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Vec<SqliteRow>, SqliteOperationError> {
if let Err(error) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
match diagnose_legacy_outcome(
SqliteOperation::TransactionRollback,
self.drain_orphaned_transaction(cx).await,
) {
Outcome::Ok(()) => {}
Outcome::Err(error) => return Outcome::Err(error),
Outcome::Cancelled(reason) => return Outcome::Cancelled(reason),
Outcome::Panicked(payload) => return Outcome::Panicked(payload),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
let sql = sql.to_string();
let params = params.to_vec();
self.run_connection_op_diagnosed(
cx,
"sqlite diagnosed query",
SqliteOperation::Step,
move |conn| {
let params_refs: Vec<&dyn rusqlite::ToSql> = params
.iter()
.map(|value| value as &dyn rusqlite::ToSql)
.collect();
let mut statement = conn.prepare_cached(&sql).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Prepare, error)
})?;
let mut rows = statement.query(params_refs.as_slice()).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Bind, error)
})?;
let mut result = Vec::new();
let mut metadata = None;
while let Some(row) = rows.next().map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Step, error)
})? {
let (column_names, columns) =
metadata.get_or_insert_with(|| sqlite_row_metadata(row));
let converted = sqlite_row_from_rusqlite_row(row, column_names, columns)
.map_err(|error| {
SqliteOperationError::from_legacy(SqliteOperation::Step, error)
})?;
result.push(converted);
}
Ok(result)
},
)
.await
}
pub async fn query_stream<'connection>(
&'connection mut self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<SqliteRowStream<'connection>, SqliteError> {
if let Err(err) = validate_checked_sql_statement(sql) {
return Outcome::Err(err);
}
self.query_stream_unchecked(cx, sql, params).await
}
pub async fn query_stream_unchecked<'connection>(
&'connection mut self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<SqliteRowStream<'connection>, SqliteError> {
if let Err(err) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(err);
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.drain_orphaned_transaction(cx).await {
Outcome::Ok(()) => {}
Outcome::Err(e) => return Outcome::Err(e),
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
let timeout =
crate::database::effective_statement_timeout(cx, self.statement_timeout_override);
if let Some(limit) = timeout {
let remaining_ns = crate::database::remaining_budget(cx)
.map_or_else(|| "none".to_string(), |d| d.as_nanos().to_string());
let base_ms = self.statement_timeout_override.map_or_else(
|| "none".to_string(),
|d| crate::database::statement_timeout_millis(d).to_string(),
);
cx.trace(&format!(
"client.budget_forwarded proto=sqlite base_ms={base_ms} \
remaining_ns={remaining_ns} statement_timeout_ms={} op=row_stream",
crate::database::statement_timeout_millis(limit)
));
}
let sql = sql.to_string();
let params: Vec<SqliteValue> = params.to_vec();
let inner = Arc::clone(&self.inner);
let counters = Arc::new(SqliteRowStreamCounters::default());
let worker_counters = Arc::clone(&counters);
let (sender, receiver) = mpsc::channel(SQLITE_ROW_STREAM_CHANNEL_CAPACITY);
let phase = Arc::new(Mutex::new(SqliteConnectionOpPhase::Queued));
let worker_phase = Arc::clone(&phase);
let handle = self.pool.spawn(move || {
const TIMEOUT_PROGRESS_OPS: i32 = 1000;
let result = (|| {
let guard = inner.lock();
{
let mut phase = worker_phase.lock();
match *phase {
SqliteConnectionOpPhase::Queued => {
*phase = SqliteConnectionOpPhase::Running;
}
SqliteConnectionOpPhase::CancelRequested => {
*phase = SqliteConnectionOpPhase::Completed;
drop(phase);
drop(guard);
return Ok(());
}
SqliteConnectionOpPhase::Running | SqliteConnectionOpPhase::Completed => {
unreachable!("a SQLite row-stream worker starts exactly once")
}
}
}
let body_result = (|| {
let conn = guard.get()?;
if let Some(limit) = timeout {
let deadline = std::time::Instant::now() + limit;
conn.progress_handler(
TIMEOUT_PROGRESS_OPS,
Some(move || std::time::Instant::now() >= deadline),
)
.map_err(|e| {
SqliteError::Sqlite(format!("failed to arm statement timeout: {e}"))
})?;
}
let query_result = (|| {
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(|v| v as &dyn rusqlite::ToSql).collect();
let mut stmt = conn
.prepare_cached(&sql)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let mut rows = stmt
.query(params_refs.as_slice())
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let mut metadata = None;
while let Some(row) = rows
.next()
.map_err(|e| SqliteError::Sqlite(e.to_string()))?
{
worker_counters.rows_stepped.fetch_add(1, Ordering::AcqRel);
let (column_names, columns) =
metadata.get_or_insert_with(|| sqlite_row_metadata(row));
let row = sqlite_row_from_rusqlite_row(row, column_names, columns)?;
if !send_sqlite_stream_message(&sender, &worker_counters, Ok(row)) {
break;
}
}
Ok(())
})();
if timeout.is_some() {
let _ = conn.progress_handler(0, None::<fn() -> bool>);
}
query_result
})();
{
let mut phase = worker_phase.lock();
*phase = SqliteConnectionOpPhase::Completed;
}
drop(guard);
body_result
})();
let result = match (timeout, result) {
(Some(limit), Err(err)) if sqlite_error_is_interrupt(&err) => {
Err(SqliteError::StatementTimeout { limit })
}
(_, result) => result,
};
if let Err(err) = result {
let _ = send_sqlite_stream_message(&sender, &worker_counters, Err(err));
}
});
Outcome::Ok(SqliteRowStream {
receiver,
handle,
counters,
phase,
finished: false,
interrupt: Arc::clone(&self.interrupt),
_connection_lease: PhantomData,
})
}
pub async fn query_row(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Option<SqliteRow>, SqliteError> {
if let Err(err) = validate_checked_sql_statement(sql) {
return Outcome::Err(err);
}
self.query_row_unchecked(cx, sql, params).await
}
pub async fn query_row_unchecked(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Option<SqliteRow>, SqliteError> {
if let Err(err) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(err);
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.drain_orphaned_transaction(cx).await {
Outcome::Ok(()) => {}
Outcome::Err(e) => return Outcome::Err(e),
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
let sql = sql.to_string();
let params: Vec<SqliteValue> = params.to_vec();
self.run_connection_op(cx, "sqlite query_row", move |conn| {
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(|v| v as &dyn rusqlite::ToSql).collect();
let mut stmt = conn
.prepare_cached(&sql)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let mut rows = stmt
.query(params_refs.as_slice())
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let row_opt = rows
.next()
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
let result = if let Some(row) = row_opt {
let (column_names, columns) = sqlite_row_metadata(row);
Some(sqlite_row_from_rusqlite_row(row, &column_names, &columns)?)
} else {
None
};
drop(rows);
drop(stmt);
Ok(result)
})
.await
}
pub async fn query_row_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Option<SqliteRow>, SqliteOperationError> {
if let Err(error) = validate_checked_sql_statement(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
self.query_row_unchecked_diagnosed(cx, sql, params).await
}
pub async fn query_row_unchecked_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Option<SqliteRow>, SqliteOperationError> {
if let Err(error) = ensure_unchecked_sql_surface(sql) {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Validation,
error,
));
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
match diagnose_legacy_outcome(
SqliteOperation::TransactionRollback,
self.drain_orphaned_transaction(cx).await,
) {
Outcome::Ok(()) => {}
Outcome::Err(error) => return Outcome::Err(error),
Outcome::Cancelled(reason) => return Outcome::Cancelled(reason),
Outcome::Panicked(payload) => return Outcome::Panicked(payload),
}
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
let sql = sql.to_string();
let params = params.to_vec();
self.run_connection_op_diagnosed(
cx,
"sqlite diagnosed query_row",
SqliteOperation::Step,
move |conn| {
let params_refs: Vec<&dyn rusqlite::ToSql> = params
.iter()
.map(|value| value as &dyn rusqlite::ToSql)
.collect();
let mut statement = conn.prepare_cached(&sql).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Prepare, error)
})?;
let mut rows = statement.query(params_refs.as_slice()).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Bind, error)
})?;
let row = rows.next().map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Step, error)
})?;
let result = match row {
Some(row) => {
let (column_names, columns) = sqlite_row_metadata(row);
Some(
sqlite_row_from_rusqlite_row(row, &column_names, &columns).map_err(
|error| {
SqliteOperationError::from_legacy(SqliteOperation::Step, error)
},
)?,
)
}
None => None,
};
Ok(result)
},
)
.await
}
async fn begin_with_sql<'conn>(
&'conn self,
cx: &Cx,
sql: &'static str,
operation: &'static str,
) -> Outcome<SqliteTransaction<'conn>, SqliteError> {
trace_database_transaction(cx, "sqlite", operation, "start");
let mut drop_guard = BeginDropGuard::new(
Arc::clone(&self.transaction_state),
Arc::clone(&self.transaction_generation),
);
let effect = TransactionWorkerEffect::Begin(drop_guard.attempt());
match self.execute_transaction_control(cx, sql, effect).await {
Outcome::Ok(_) => {
let Some(generation) = drop_guard.opened_generation() else {
drop_guard.abandon();
trace_database_transaction(cx, "sqlite", operation, "err");
return Outcome::Err(SqliteError::Sqlite(
"managed BEGIN completed without a transaction generation".to_string(),
));
};
let transaction = SqliteTransaction {
conn: self,
finished: false,
obligation: reserve_transaction_obligation(cx),
generation,
};
drop_guard.disarm();
trace_database_transaction(cx, "sqlite", operation, "ok");
Outcome::Ok(transaction)
}
Outcome::Err(e) => {
drop_guard.disarm();
trace_database_transaction(cx, "sqlite", operation, "err");
Outcome::Err(e)
}
Outcome::Cancelled(r) => {
drop_guard.abandon();
trace_database_transaction(cx, "sqlite", operation, "cancelled");
Outcome::Cancelled(r)
}
Outcome::Panicked(p) => {
drop_guard.abandon();
trace_database_transaction(cx, "sqlite", operation, "panicked");
Outcome::Panicked(p)
}
}
}
async fn begin_with_sql_diagnosed<'conn>(
&'conn self,
cx: &Cx,
sql: &'static str,
trace_operation: &'static str,
) -> Outcome<SqliteTransaction<'conn>, SqliteOperationError> {
trace_database_transaction(cx, "sqlite", trace_operation, "start");
let mut drop_guard = BeginDropGuard::new(
Arc::clone(&self.transaction_state),
Arc::clone(&self.transaction_generation),
);
let effect = TransactionWorkerEffect::Begin(drop_guard.attempt());
match self
.execute_transaction_control_diagnosed(
cx,
sql,
SqliteOperation::TransactionBegin,
effect,
)
.await
{
Outcome::Ok(_) => {
let Some(generation) = drop_guard.opened_generation() else {
drop_guard.abandon();
trace_database_transaction(cx, "sqlite", trace_operation, "err");
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::TransactionBegin,
SqliteError::Sqlite(
"managed BEGIN completed without a transaction generation".to_string(),
),
));
};
let transaction = SqliteTransaction {
conn: self,
finished: false,
obligation: reserve_transaction_obligation(cx),
generation,
};
drop_guard.disarm();
trace_database_transaction(cx, "sqlite", trace_operation, "ok");
Outcome::Ok(transaction)
}
Outcome::Err(error) => {
drop_guard.disarm();
trace_database_transaction(cx, "sqlite", trace_operation, "err");
Outcome::Err(error)
}
Outcome::Cancelled(reason) => {
drop_guard.abandon();
trace_database_transaction(cx, "sqlite", trace_operation, "cancelled");
Outcome::Cancelled(reason)
}
Outcome::Panicked(payload) => {
drop_guard.abandon();
trace_database_transaction(cx, "sqlite", trace_operation, "panicked");
Outcome::Panicked(payload)
}
}
}
pub async fn begin(&self, cx: &Cx) -> Outcome<SqliteTransaction<'_>, SqliteError> {
self.begin_with_sql(cx, "BEGIN", "begin").await
}
pub async fn begin_diagnosed(
&self,
cx: &Cx,
) -> Outcome<SqliteTransaction<'_>, SqliteOperationError> {
self.begin_with_sql_diagnosed(cx, "BEGIN", "begin_diagnosed")
.await
}
pub async fn begin_immediate(&self, cx: &Cx) -> Outcome<SqliteTransaction<'_>, SqliteError> {
self.begin_with_sql(cx, "BEGIN IMMEDIATE", "begin_immediate")
.await
}
pub async fn begin_immediate_diagnosed(
&self,
cx: &Cx,
) -> Outcome<SqliteTransaction<'_>, SqliteOperationError> {
self.begin_with_sql_diagnosed(cx, "BEGIN IMMEDIATE", "begin_immediate_diagnosed")
.await
}
pub async fn begin_exclusive(&self, cx: &Cx) -> Outcome<SqliteTransaction<'_>, SqliteError> {
self.begin_with_sql(cx, "BEGIN EXCLUSIVE", "begin_exclusive")
.await
}
pub async fn begin_exclusive_diagnosed(
&self,
cx: &Cx,
) -> Outcome<SqliteTransaction<'_>, SqliteOperationError> {
self.begin_with_sql_diagnosed(cx, "BEGIN EXCLUSIVE", "begin_exclusive_diagnosed")
.await
}
pub async fn set_busy_timeout(&self, cx: &Cx, timeout: Duration) -> Outcome<(), SqliteError> {
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.drain_orphaned_transaction(cx).await {
Outcome::Ok(()) => {}
Outcome::Err(e) => return Outcome::Err(e),
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
self.run_connection_op(cx, "sqlite set_busy_timeout", move |conn| {
conn.busy_timeout(timeout)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
Ok(())
})
.await
}
pub async fn set_busy_timeout_diagnosed(
&self,
cx: &Cx,
timeout: Duration,
) -> Outcome<(), SqliteOperationError> {
if cx.checkpoint().is_err() {
return Outcome::Cancelled(sqlite_cancelled_reason(cx));
}
match diagnose_legacy_outcome(
SqliteOperation::TransactionRollback,
self.drain_orphaned_transaction(cx).await,
) {
Outcome::Ok(()) => {}
Outcome::Err(error) => return Outcome::Err(error),
Outcome::Cancelled(reason) => return Outcome::Cancelled(reason),
Outcome::Panicked(payload) => return Outcome::Panicked(payload),
}
self.run_connection_op_diagnosed(
cx,
"sqlite diagnosed set_busy_timeout",
SqliteOperation::Configure,
move |conn| {
conn.busy_timeout(timeout).map_err(|error| {
SqliteOperationError::from_rusqlite(SqliteOperation::Configure, error)
})
},
)
.await
}
pub fn close(&self) -> Result<(), SqliteError> {
let mut guard = self.inner.lock();
if let Some(conn) = guard.conn.as_ref() {
let _ =
rollback_orphaned_transaction_mutex_guarded(conn, self.transaction_state.as_ref());
match self.execute_wal_checkpoint_with_retry(conn) {
Ok(()) => {
#[cfg(feature = "tracing-integration")]
crate::tracing_compat::debug!(
"WAL checkpoint completed successfully during close"
);
}
Err(e) => {
#[cfg(feature = "tracing-integration")]
crate::tracing_compat::error!(
error = %e,
"WAL checkpoint failed during connection close - failing close to prevent data loss"
);
return Err(e);
}
}
conn.flush_prepared_statement_cache();
}
*self.transaction_state.lock() = TransactionState::Autocommit;
guard.close();
Ok(())
}
pub fn close_diagnosed(&self) -> Result<(), SqliteOperationError> {
self.close()
.map_err(|error| SqliteOperationError::from_legacy(SqliteOperation::Close, error))
}
pub async fn close_async(&self, cx: &Cx) -> Outcome<(), SqliteError> {
if cx.checkpoint().is_err() {
return Outcome::Cancelled(
cx.cancel_reason()
.unwrap_or_else(|| CancelReason::user("cancelled")),
);
}
match self.execute_wal_checkpoint_async_with_retry(cx).await {
Outcome::Ok(()) => {
#[cfg(feature = "tracing-integration")]
crate::tracing_compat::debug!("Async WAL checkpoint completed successfully");
}
Outcome::Err(e) => {
#[cfg(feature = "tracing-integration")]
crate::tracing_compat::error!(
error = %e,
"Async WAL checkpoint failed during connection close - failing close to prevent data loss"
);
return Outcome::Err(e);
}
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
match self.close_without_checkpoint() {
Ok(()) => Outcome::Ok(()),
Err(e) => Outcome::Err(e),
}
}
pub async fn close_async_diagnosed(&self, cx: &Cx) -> Outcome<(), SqliteOperationError> {
diagnose_legacy_outcome(SqliteOperation::Close, self.close_async(cx).await)
}
#[must_use]
pub fn is_open(&self) -> bool {
self.inner.lock().conn.is_some()
}
fn execute_wal_checkpoint_with_retry(
&self,
conn: &rusqlite::Connection,
) -> Result<(), SqliteError> {
const MAX_RETRY_ATTEMPTS: u32 = 3;
const RETRY_DELAY_MS: u64 = 50;
for attempt in 1..=MAX_RETRY_ATTEMPTS {
match self.execute_single_wal_checkpoint(conn) {
Ok(()) => {
#[cfg(feature = "tracing-integration")]
if attempt > 1 {
crate::tracing_compat::info!(
attempt = attempt,
"WAL checkpoint succeeded after retry"
);
}
return Ok(());
}
Err(e) => {
#[cfg(feature = "tracing-integration")]
crate::tracing_compat::warn!(
error = %e,
attempt = attempt,
max_attempts = MAX_RETRY_ATTEMPTS,
"WAL checkpoint attempt failed"
);
if attempt == MAX_RETRY_ATTEMPTS {
return Err(SqliteError::WalCheckpointFailed(format!(
"WAL checkpoint failed after {} attempts: {}",
MAX_RETRY_ATTEMPTS, e
)));
}
std::thread::sleep(std::time::Duration::from_millis(
RETRY_DELAY_MS * attempt as u64,
));
}
}
}
unreachable!("Loop should always return within max attempts")
}
fn execute_single_wal_checkpoint(
&self,
conn: &rusqlite::Connection,
) -> Result<(), rusqlite::Error> {
conn.execute_batch("PRAGMA wal_checkpoint(RESTART)")?;
let mut stmt = conn.prepare_cached("PRAGMA wal_checkpoint")?;
let result: (i32, i32, i32) =
stmt.query_row([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?;
let (busy, log_pages, checkpointed_pages) = result;
if busy != 0 {
return Err(rusqlite::Error::SqliteFailure(
rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
Some("WAL checkpoint blocked by concurrent readers".to_string()),
));
}
if log_pages > 0 && checkpointed_pages == 0 {
return Err(rusqlite::Error::SqliteFailure(
rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR),
Some(format!(
"WAL checkpoint failed - {} pages remain in WAL",
log_pages
)),
));
}
Ok(())
}
async fn execute_wal_checkpoint_async_with_retry(&self, cx: &Cx) -> Outcome<(), SqliteError> {
const MAX_RETRY_ATTEMPTS: u32 = 3;
for attempt in 1..=MAX_RETRY_ATTEMPTS {
match self.execute_wal_checkpoint_async_single(cx).await {
Outcome::Ok(()) => {
#[cfg(feature = "tracing-integration")]
if attempt > 1 {
crate::tracing_compat::info!(
attempt = attempt,
"Async WAL checkpoint succeeded after retry"
);
}
return Outcome::Ok(());
}
Outcome::Err(e) => {
#[cfg(feature = "tracing-integration")]
crate::tracing_compat::warn!(
error = %e,
attempt = attempt,
max_attempts = MAX_RETRY_ATTEMPTS,
"Async WAL checkpoint attempt failed"
);
if attempt == MAX_RETRY_ATTEMPTS {
return Outcome::Err(SqliteError::WalCheckpointFailed(format!(
"Async WAL checkpoint failed after {} attempts: {}",
MAX_RETRY_ATTEMPTS, e
)));
}
let retry_delay = Duration::from_millis(50 * u64::from(attempt));
if let Err(reason) = sqlite_wait_retry_delay(cx, retry_delay).await {
return Outcome::Cancelled(reason);
}
}
Outcome::Cancelled(r) => return Outcome::Cancelled(r),
Outcome::Panicked(p) => return Outcome::Panicked(p),
}
}
unreachable!("Loop should always return within max attempts")
}
async fn execute_wal_checkpoint_async_single(&self, cx: &Cx) -> Outcome<(), SqliteError> {
match self
.execute_batch_unchecked(cx, "PRAGMA wal_checkpoint(RESTART)")
.await
{
Outcome::Ok(()) => {
match self.query_unchecked(cx, "PRAGMA wal_checkpoint", &[]).await {
Outcome::Ok(rows) => {
if let Some(row) = rows.first() {
let busy = match wal_checkpoint_i64(row, "busy") {
Ok(value) => value,
Err(err) => return Outcome::Err(err),
};
let log_pages = match wal_checkpoint_i64(row, "log") {
Ok(value) => value,
Err(err) => return Outcome::Err(err),
};
let checkpointed_pages = match wal_checkpoint_i64(row, "checkpointed") {
Ok(value) => value,
Err(err) => return Outcome::Err(err),
};
if busy != 0 {
return Outcome::Err(SqliteError::WalCheckpointFailed(
"WAL checkpoint blocked by concurrent readers".to_string(),
));
}
if log_pages > 0 && checkpointed_pages == 0 {
return Outcome::Err(SqliteError::WalCheckpointFailed(format!(
"WAL checkpoint failed - {} pages remain in WAL",
log_pages
)));
}
}
Outcome::Ok(())
}
Outcome::Err(e) => Outcome::Err(e),
Outcome::Cancelled(r) => Outcome::Cancelled(r),
Outcome::Panicked(p) => Outcome::Panicked(p),
}
}
Outcome::Err(e) => Outcome::Err(e),
Outcome::Cancelled(r) => Outcome::Cancelled(r),
Outcome::Panicked(p) => Outcome::Panicked(p),
}
}
fn close_without_checkpoint(&self) -> Result<(), SqliteError> {
let mut guard = self.inner.lock();
if let Some(conn) = guard.conn.as_ref() {
let _ =
rollback_orphaned_transaction_mutex_guarded(conn, self.transaction_state.as_ref());
conn.flush_prepared_statement_cache();
}
*self.transaction_state.lock() = TransactionState::Autocommit;
guard.close();
Ok(())
}
}
pub struct SqliteTransaction<'a> {
conn: &'a SqliteConnection,
finished: bool,
obligation: Option<ObligationToken<TransactionKind>>,
generation: u64,
}
fn reserve_transaction_obligation(cx: &Cx) -> Option<ObligationToken<TransactionKind>> {
let region = cx.region_id();
if region.as_u64() == 0 {
None
} else {
Some(ObligationToken::reserve("db-transaction:sqlite", region))
}
}
impl SqliteTransaction<'_> {
#[must_use]
pub(crate) fn requires_rollback_before_commit(&self) -> bool {
let state = self.conn.transaction_state.lock();
self.conn.transaction_generation.load(Ordering::Acquire) == self.generation
&& *state == TransactionState::NeedsRollback
}
pub(crate) fn poison_for_rollback(&self) {
let mut state = self.conn.transaction_state.lock();
if self.conn.transaction_generation.load(Ordering::Acquire) == self.generation {
*state = TransactionState::NeedsRollback;
}
}
pub async fn commit(mut self, cx: &Cx) -> Outcome<(), SqliteError> {
if self.finished {
trace_database_transaction(cx, "sqlite", "commit", "already_finished");
return Outcome::Err(SqliteError::TransactionFinished);
}
trace_database_transaction(cx, "sqlite", "commit", "start");
let finish_effect = TransactionFinishEffect::new(
Arc::clone(&self.conn.transaction_state),
Arc::clone(&self.conn.transaction_generation),
self.generation,
TransactionFinishKind::Commit,
self.obligation.take(),
);
let effect = TransactionWorkerEffect::Finish(finish_effect);
match self
.conn
.execute_transaction_control(cx, "COMMIT", effect)
.await
{
Outcome::Ok(_) => {
self.finished = true;
trace_database_transaction(cx, "sqlite", "commit", "ok");
Outcome::Ok(())
}
Outcome::Err(e) => {
trace_database_transaction(cx, "sqlite", "commit", "err");
Outcome::Err(e)
}
Outcome::Cancelled(r) => {
trace_database_transaction(cx, "sqlite", "commit", "cancelled");
Outcome::Cancelled(r)
}
Outcome::Panicked(p) => {
trace_database_transaction(cx, "sqlite", "commit", "panicked");
Outcome::Panicked(p)
}
}
}
pub async fn commit_diagnosed(mut self, cx: &Cx) -> Outcome<(), SqliteOperationError> {
if self.finished {
trace_database_transaction(cx, "sqlite", "commit_diagnosed", "already_finished");
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::TransactionCommit,
SqliteError::TransactionFinished,
));
}
trace_database_transaction(cx, "sqlite", "commit_diagnosed", "start");
let finish_effect = TransactionFinishEffect::new(
Arc::clone(&self.conn.transaction_state),
Arc::clone(&self.conn.transaction_generation),
self.generation,
TransactionFinishKind::Commit,
self.obligation.take(),
);
let effect = TransactionWorkerEffect::Finish(finish_effect);
match self
.conn
.execute_transaction_control_diagnosed(
cx,
"COMMIT",
SqliteOperation::TransactionCommit,
effect,
)
.await
{
Outcome::Ok(_) => {
self.finished = true;
trace_database_transaction(cx, "sqlite", "commit_diagnosed", "ok");
Outcome::Ok(())
}
Outcome::Err(error) => {
trace_database_transaction(cx, "sqlite", "commit_diagnosed", "err");
Outcome::Err(error)
}
Outcome::Cancelled(reason) => {
trace_database_transaction(cx, "sqlite", "commit_diagnosed", "cancelled");
Outcome::Cancelled(reason)
}
Outcome::Panicked(payload) => {
trace_database_transaction(cx, "sqlite", "commit_diagnosed", "panicked");
Outcome::Panicked(payload)
}
}
}
pub async fn rollback(mut self, cx: &Cx) -> Outcome<(), SqliteError> {
if self.finished {
trace_database_transaction(cx, "sqlite", "rollback", "already_finished");
return Outcome::Err(SqliteError::TransactionFinished);
}
trace_database_transaction(cx, "sqlite", "rollback", "start");
let finish_effect = TransactionFinishEffect::new(
Arc::clone(&self.conn.transaction_state),
Arc::clone(&self.conn.transaction_generation),
self.generation,
TransactionFinishKind::Rollback,
self.obligation.take(),
);
let effect = TransactionWorkerEffect::Finish(finish_effect);
match self
.conn
.execute_transaction_control(cx, "ROLLBACK", effect)
.await
{
Outcome::Ok(_) => {
self.finished = true;
trace_database_transaction(cx, "sqlite", "rollback", "ok");
Outcome::Ok(())
}
Outcome::Err(e) => {
trace_database_transaction(cx, "sqlite", "rollback", "err");
Outcome::Err(e)
}
Outcome::Cancelled(r) => {
trace_database_transaction(cx, "sqlite", "rollback", "cancelled");
Outcome::Cancelled(r)
}
Outcome::Panicked(p) => {
trace_database_transaction(cx, "sqlite", "rollback", "panicked");
Outcome::Panicked(p)
}
}
}
pub async fn rollback_diagnosed(mut self, cx: &Cx) -> Outcome<(), SqliteOperationError> {
if self.finished {
trace_database_transaction(cx, "sqlite", "rollback_diagnosed", "already_finished");
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::TransactionRollback,
SqliteError::TransactionFinished,
));
}
trace_database_transaction(cx, "sqlite", "rollback_diagnosed", "start");
let finish_effect = TransactionFinishEffect::new(
Arc::clone(&self.conn.transaction_state),
Arc::clone(&self.conn.transaction_generation),
self.generation,
TransactionFinishKind::Rollback,
self.obligation.take(),
);
let effect = TransactionWorkerEffect::Finish(finish_effect);
match self
.conn
.execute_transaction_control_diagnosed(
cx,
"ROLLBACK",
SqliteOperation::TransactionRollback,
effect,
)
.await
{
Outcome::Ok(_) => {
self.finished = true;
trace_database_transaction(cx, "sqlite", "rollback_diagnosed", "ok");
Outcome::Ok(())
}
Outcome::Err(error) => {
trace_database_transaction(cx, "sqlite", "rollback_diagnosed", "err");
Outcome::Err(error)
}
Outcome::Cancelled(reason) => {
trace_database_transaction(cx, "sqlite", "rollback_diagnosed", "cancelled");
Outcome::Cancelled(reason)
}
Outcome::Panicked(payload) => {
trace_database_transaction(cx, "sqlite", "rollback_diagnosed", "panicked");
Outcome::Panicked(payload)
}
}
}
pub async fn execute(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteError> {
if self.finished {
return Outcome::Err(SqliteError::TransactionFinished);
}
self.conn.execute(cx, sql, params).await
}
pub async fn execute_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteOperationError> {
if self.finished {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Step,
SqliteError::TransactionFinished,
));
}
self.conn.execute_diagnosed(cx, sql, params).await
}
pub(crate) async fn execute_unchecked(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<u64, SqliteError> {
if self.finished {
return Outcome::Err(SqliteError::TransactionFinished);
}
self.conn.execute_unchecked(cx, sql, params).await
}
pub async fn query(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Vec<SqliteRow>, SqliteError> {
if self.finished {
return Outcome::Err(SqliteError::TransactionFinished);
}
self.conn.query(cx, sql, params).await
}
pub async fn query_diagnosed(
&self,
cx: &Cx,
sql: &str,
params: &[SqliteValue],
) -> Outcome<Vec<SqliteRow>, SqliteOperationError> {
if self.finished {
return Outcome::Err(SqliteOperationError::from_legacy(
SqliteOperation::Step,
SqliteError::TransactionFinished,
));
}
self.conn.query_diagnosed(cx, sql, params).await
}
}
impl Drop for SqliteTransaction<'_> {
fn drop(&mut self) {
if let Some(token) = self.obligation.take() {
let _ = token.abort();
}
if !self.finished {
self.poison_for_rollback();
self.conn
.schedule_dropped_transaction_rollback(self.generation);
}
}
}
fn column_name_or_index(column_names: &[String], idx: usize) -> String {
column_names
.get(idx)
.cloned()
.unwrap_or_else(|| format!("index {idx}"))
}
fn convert_value(
value: rusqlite::types::ValueRef<'_>,
column: &str,
) -> Result<SqliteValue, SqliteError> {
match value {
rusqlite::types::ValueRef::Null => Ok(SqliteValue::Null),
rusqlite::types::ValueRef::Integer(v) => Ok(SqliteValue::Integer(v)),
rusqlite::types::ValueRef::Real(v) => Ok(SqliteValue::Real(v)),
rusqlite::types::ValueRef::Text(v) => {
let text =
std::str::from_utf8(v).map_err(|source| SqliteError::InvalidTextEncoding {
column: column.to_string(),
source,
})?;
Ok(SqliteValue::Text(text.to_string()))
}
rusqlite::types::ValueRef::Blob(v) => Ok(SqliteValue::Blob(v.to_vec())),
}
}
impl rusqlite::ToSql for SqliteValue {
fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
use rusqlite::types::ToSqlOutput;
match self {
Self::Null => Ok(ToSqlOutput::Owned(rusqlite::types::Value::Null)),
Self::Integer(v) => Ok(ToSqlOutput::Owned(rusqlite::types::Value::Integer(*v))),
Self::Real(v) => Ok(ToSqlOutput::Owned(rusqlite::types::Value::Real(*v))),
Self::Text(v) => Ok(ToSqlOutput::Owned(rusqlite::types::Value::Text(v.clone()))),
Self::Blob(v) => Ok(ToSqlOutput::Owned(rusqlite::types::Value::Blob(v.clone()))),
}
}
}
#[cfg(test)]
mod tests {
#![allow(
clippy::pedantic,
clippy::nursery,
clippy::expect_fun_call,
clippy::map_unwrap_or,
clippy::cast_possible_wrap,
clippy::future_not_send
)]
use super::*;
use crate::conformance::{ConformanceTarget, LabRuntimeTarget, TestConfig};
use crate::cx::Cx;
use crate::test_utils::init_test_logging;
use crate::types::Budget;
use crate::types::Outcome;
use crate::util::ArenaIndex;
use crate::{RegionId, TaskId};
use futures_lite::future::block_on;
use tempfile::tempdir;
#[test]
fn test_sqlparser_blocks_pragma() {
assert_eq!(
classify_sql_surface_violation("PRAGMA journal_mode"),
Some(SqlSurfaceViolation::Pragma)
);
assert_eq!(
classify_sql_surface_violation("pragma foreign_keys"),
Some(SqlSurfaceViolation::Pragma)
);
assert_eq!(
classify_sql_surface_violation("/* comment */ PRAGMA journal_mode"),
Some(SqlSurfaceViolation::Pragma)
);
assert_eq!(classify_sql_surface_violation("SELECT * FROM users"), None);
assert_eq!(
classify_sql_surface_violation("INSERT INTO test VALUES (1, 'test')"),
None
);
}
#[test]
fn test_sqlparser_blocks_attach_detach() {
assert_eq!(
classify_sql_surface_violation("ATTACH 'db.sqlite' AS test"),
Some(SqlSurfaceViolation::AttachDetach)
);
assert_eq!(
classify_sql_surface_violation("DETACH DATABASE test"),
Some(SqlSurfaceViolation::AttachDetach)
);
assert_eq!(classify_sql_surface_violation("SELECT * FROM users"), None);
}
#[test]
fn test_sqlparser_blocks_transaction_control() {
assert_eq!(
classify_sql_surface_violation("BEGIN IMMEDIATE"),
Some(SqlSurfaceViolation::TransactionControl)
);
assert_eq!(
classify_sql_surface_violation("COMMIT"),
Some(SqlSurfaceViolation::TransactionControl)
);
assert_eq!(
classify_sql_surface_violation("ROLLBACK"),
Some(SqlSurfaceViolation::TransactionControl)
);
assert_eq!(
classify_sql_surface_violation(
"CREATE TRIGGER test AFTER INSERT ON table BEGIN INSERT INTO log VALUES (1); END"
),
None
);
assert_eq!(classify_sql_surface_violation("SELECT * FROM users"), None);
}
#[test]
fn test_checked_sql_blocks_extension_loading_calls() {
for sql in [
"SELECT load_extension('/tmp/evil.so')",
"SELECT LOAD_EXTENSION ( '/tmp/evil.so', 'entrypoint' )",
"SELECT main.load_extension('/tmp/evil.so')",
"SELECT main.\"load_extension\"('/tmp/evil.so')",
"SELECT \"load_extension\"('/tmp/evil.so')",
"SELECT [load_extension]('/tmp/evil.so')",
"SELECT `load_extension`('/tmp/evil.so')",
"SELECT load_extension /* comment */ ('/tmp/evil.so')",
] {
assert_eq!(
classify_sql_surface_violation(sql),
Some(SqlSurfaceViolation::ExtensionLoading),
"checked policy must reject {sql:?}"
);
}
assert_eq!(
classify_sql_surface_violation("SELECT 'load_extension(' AS inert_text"),
None,
"extension-like text inside a string literal is data"
);
}
#[test]
fn test_sqlparser_comment_bypass_protection() {
let sql = "/* comment */ PRAGMA journal_mode -- line comment";
assert_eq!(
classify_sql_surface_violation(sql),
Some(SqlSurfaceViolation::Pragma)
);
let sql = "/* outer /* inner */ comment */ SELECT 1";
assert_eq!(
classify_sql_surface_violation(sql),
Some(SqlSurfaceViolation::ParserRejected)
);
}
#[test]
fn test_toctou_fix_path_resolution() {
use std::fs;
use tempfile::tempdir;
let temp_dir = tempdir().expect("Failed to create temp directory");
let temp_path = temp_dir.path();
let db_file = temp_path.join("test.sqlite");
fs::write(&db_file, b"").expect("Failed to create test database file");
let resolved = resolve_sqlite_open_path(&db_file).expect("Failed to resolve path");
validate_resolved_sqlite_path(&resolved).expect("Safe path should validate");
let etc_path = Path::new("/etc/passwd");
assert!(validate_resolved_sqlite_path(etc_path).is_err());
let sys_path = Path::new("/sys/kernel");
assert!(validate_resolved_sqlite_path(sys_path).is_err());
let proc_path = Path::new("/proc/version");
assert!(validate_resolved_sqlite_path(proc_path).is_err());
let dev_path = Path::new("/dev/null");
assert!(validate_resolved_sqlite_path(dev_path).is_err());
}
#[test]
#[cfg(unix)]
fn test_toctou_fix_prevents_symlink_attack() {
use std::os::unix::fs::symlink;
use tempfile::tempdir;
let temp_dir = tempdir().expect("Failed to create temp directory");
let temp_path = temp_dir.path();
let symlink_path = temp_path.join("malicious.sqlite");
if symlink("/etc/passwd", &symlink_path).is_ok() {
let resolved =
resolve_sqlite_open_path(&symlink_path).expect("Failed to resolve symlink");
assert!(validate_resolved_sqlite_path(&resolved).is_err());
assert!(resolved.starts_with("/etc"));
}
}
#[test]
fn test_path_validation_comprehensive() {
let tilde_path = Path::new("~/database.sqlite");
assert!(validate_sqlite_open_path(tilde_path).is_err());
let traversal_path = Path::new("../../../etc/passwd");
assert!(validate_sqlite_open_path(traversal_path).is_err());
let current_path = Path::new("./test.sqlite");
let _ = validate_sqlite_open_path(current_path);
}
#[test]
fn test_wal_checkpoint_fail_closed() {
use tempfile::NamedTempFile;
let _temp_file = NamedTempFile::new().expect("Failed to create temp file");
let checkpoint_error = SqliteError::WalCheckpointFailed("test error".to_string());
assert!(matches!(
checkpoint_error,
SqliteError::WalCheckpointFailed(_)
));
let error_msg = format!("{}", checkpoint_error);
assert!(error_msg.contains("WAL checkpoint failed"));
assert!(error_msg.contains("test error"));
}
#[test]
fn test_wal_checkpoint_error_variants() {
let busy_error = SqliteError::WalCheckpointFailed(
"WAL checkpoint blocked by concurrent readers".to_string(),
);
assert!(format!("{}", busy_error).contains("blocked by concurrent readers"));
let incomplete_error = SqliteError::WalCheckpointFailed(
"WAL checkpoint failed - 42 pages remain in WAL".to_string(),
);
assert!(format!("{}", incomplete_error).contains("pages remain in WAL"));
let retry_error = SqliteError::WalCheckpointFailed(
"WAL checkpoint failed after 3 attempts: I/O error".to_string(),
);
assert!(format!("{}", retry_error).contains("failed after 3 attempts"));
}
#[test]
fn test_wal_checkpoint_security_properties() {
let checkpoint_failure = SqliteError::WalCheckpointFailed("simulated failure".to_string());
assert!(matches!(
checkpoint_failure,
SqliteError::WalCheckpointFailed(_)
));
const MAX_RETRY_ATTEMPTS: u32 = 3;
assert_eq!(MAX_RETRY_ATTEMPTS, 3);
let restart_pragma = "PRAGMA wal_checkpoint(RESTART)";
assert!(restart_pragma.contains("RESTART"));
assert!(!restart_pragma.contains("FULL"));
}
#[test]
fn test_mutex_transaction_state_transitions() {
let transaction_state = Mutex::new(TransactionState::Autocommit);
{
let mut guard = transaction_state.lock();
*guard = TransactionState::InTransaction;
}
assert_eq!(*transaction_state.lock(), TransactionState::InTransaction);
{
let mut guard = transaction_state.lock();
assert_eq!(*guard, TransactionState::InTransaction);
*guard = TransactionState::NeedsRollback;
}
assert_eq!(*transaction_state.lock(), TransactionState::NeedsRollback);
}
#[test]
fn test_concurrent_rollback_prevention() {
use std::sync::Arc;
use std::thread;
let transaction_state = Arc::new(Mutex::new(TransactionState::NeedsRollback));
let state1 = Arc::clone(&transaction_state);
let state2 = Arc::clone(&transaction_state);
let handle1 = thread::spawn(move || {
let mut guard = state1.lock();
if *guard == TransactionState::NeedsRollback {
*guard = TransactionState::RollingBack;
true } else {
false
}
});
let handle2 = thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_nanos(1));
let mut guard = state2.lock();
if *guard == TransactionState::NeedsRollback {
*guard = TransactionState::RollingBack;
true
} else {
false }
});
let result1 = handle1.join().unwrap();
let result2 = handle2.join().unwrap();
assert_ne!(
result1, result2,
"Mutex should prevent concurrent state modification"
);
assert_eq!(*transaction_state.lock(), TransactionState::RollingBack);
}
#[test]
fn test_rollback_state_machine() {
let transaction_state = Mutex::new(TransactionState::Autocommit);
{
let mut guard = transaction_state.lock();
*guard = TransactionState::InTransaction;
}
assert_eq!(*transaction_state.lock(), TransactionState::InTransaction);
{
let mut guard = transaction_state.lock();
*guard = TransactionState::NeedsRollback;
}
assert_eq!(*transaction_state.lock(), TransactionState::NeedsRollback);
{
let mut guard = transaction_state.lock();
if *guard == TransactionState::NeedsRollback {
*guard = TransactionState::RollingBack;
}
}
assert_eq!(*transaction_state.lock(), TransactionState::RollingBack);
{
let mut guard = transaction_state.lock();
*guard = TransactionState::Autocommit;
}
assert_eq!(*transaction_state.lock(), TransactionState::Autocommit);
}
#[test]
fn test_concurrency_race_conditions_fixed() {
let transaction_state = Mutex::new(TransactionState::Autocommit);
{
let mut guard = transaction_state.lock();
*guard = TransactionState::NeedsRollback;
}
assert_eq!(*transaction_state.lock(), TransactionState::NeedsRollback);
assert_ne!(
TransactionState::RollingBack,
TransactionState::NeedsRollback
); }
#[test]
fn test_parking_lot_mutex_guard_scoping() {
use std::sync::Arc;
use std::thread;
let transaction_state = Arc::new(Mutex::new(TransactionState::Autocommit));
let state_for_thread = Arc::clone(&transaction_state);
{
let mut guard = transaction_state.lock();
*guard = TransactionState::InTransaction;
}
let handle = thread::spawn(move || {
let mut guard = state_for_thread.lock();
assert_eq!(*guard, TransactionState::InTransaction);
*guard = TransactionState::NeedsRollback;
});
handle.join().unwrap();
assert_eq!(*transaction_state.lock(), TransactionState::NeedsRollback);
}
#[test]
fn test_rollback_mutex_synchronization() {
let conn = rusqlite::Connection::open_in_memory().unwrap();
let transaction_state = Mutex::new(TransactionState::NeedsRollback);
let result = rollback_orphaned_transaction_mutex_guarded(&conn, &transaction_state);
assert!(result.is_ok());
assert_eq!(*transaction_state.lock(), TransactionState::Autocommit);
}
fn create_test_cx() -> Cx {
Cx::new(
RegionId::from_arena(ArenaIndex::new(0, 1)),
TaskId::from_arena(ArenaIndex::new(0, 0)),
Budget::INFINITE,
)
}
const INFINITE_QUERY: &str =
"WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c) SELECT count(*) FROM c";
fn run_signalled_infinite_query(
conn: &rusqlite::Connection,
started: std::sync::mpsc::SyncSender<()>,
) -> Result<(), SqliteError> {
const RUNNING_PROGRESS_CALLBACKS: usize = 10_000;
let mut progress_callbacks = 0usize;
conn.progress_handler(
1,
Some(move || {
progress_callbacks = progress_callbacks.saturating_add(1);
if progress_callbacks == RUNNING_PROGRESS_CALLBACKS {
let _ = started.try_send(());
}
false
}),
)
.map_err(|error| SqliteError::Sqlite(error.to_string()))?;
let result = conn
.query_row(INFINITE_QUERY, [], |_| Ok(()))
.map_err(|error| SqliteError::Sqlite(error.to_string()));
let _ = conn.progress_handler(0, None::<fn() -> bool>);
result
}
fn run_signalled_infinite_query_diagnosed(
conn: &rusqlite::Connection,
started: std::sync::mpsc::SyncSender<()>,
) -> Result<(), SqliteOperationError> {
const RUNNING_PROGRESS_CALLBACKS: usize = 10_000;
let mut progress_callbacks = 0usize;
conn.progress_handler(
1,
Some(move || {
progress_callbacks = progress_callbacks.saturating_add(1);
if progress_callbacks == RUNNING_PROGRESS_CALLBACKS {
let _ = started.try_send(());
}
false
}),
)
.map_err(|error| SqliteOperationError::from_rusqlite(SqliteOperation::Step, error))?;
let result = conn
.query_row(INFINITE_QUERY, [], |_| Ok(()))
.map_err(|error| SqliteOperationError::from_rusqlite(SqliteOperation::Step, error));
let _ = conn.progress_handler(0, None::<fn() -> bool>);
result
}
fn traced_cx_with_budget(budget: Budget) -> (Cx, crate::trace::TraceBufferHandle) {
let cx = Cx::new(
RegionId::from_arena(ArenaIndex::new(0, 1)),
TaskId::from_arena(ArenaIndex::new(0, 0)),
budget,
);
let trace = crate::trace::TraceBufferHandle::new(64);
cx.set_trace_buffer(trace.clone());
(cx, trace)
}
fn user_trace_messages(trace: &crate::trace::TraceBufferHandle, prefix: &str) -> Vec<String> {
trace
.snapshot()
.iter()
.filter(|e| e.kind == crate::trace::TraceEventKind::UserTrace)
.filter_map(|e| match &e.data {
crate::trace::TraceData::Message(msg) if msg.starts_with(prefix) => {
Some(msg.clone())
}
_ => None,
})
.collect()
}
#[test]
fn statement_timeout_override_aborts_runaway_query() {
init_test_logging();
let (cx, trace) = traced_cx_with_budget(Budget::INFINITE);
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
conn.set_statement_timeout_override(Some(Duration::from_millis(50)));
match conn.query_unchecked(&cx, INFINITE_QUERY, &[]).await {
Outcome::Err(SqliteError::StatementTimeout { limit }) => {
assert_eq!(limit, Duration::from_millis(50));
}
other => panic!("expected StatementTimeout, got {other:?}"),
}
match conn.query_unchecked(&cx, "SELECT 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after statement timeout: {other:?}"),
}
});
let forwarded = user_trace_messages(&trace, "client.budget_forwarded proto=sqlite ");
assert!(
forwarded
.iter()
.any(|m| m.contains("base_ms=50") && m.contains("statement_timeout_ms=50")),
"expected forwarded budget trace, got {forwarded:?}"
);
}
#[test]
fn budget_deadline_aborts_runaway_query() {
init_test_logging();
let now = crate::time::wall_now();
let (cx, trace) = traced_cx_with_budget(
Budget::INFINITE.tightened_by_timeout(now, Duration::from_millis(150)),
);
let timed_out = block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn.query_unchecked(&cx, INFINITE_QUERY, &[]).await {
Outcome::Err(SqliteError::StatementTimeout { limit }) => {
assert!(
limit <= Duration::from_millis(150),
"budget-derived limit must not exceed the remaining budget, got {limit:?}"
);
true
}
Outcome::Cancelled(_) => false,
other => panic!("expected StatementTimeout or Cancelled, got {other:?}"),
}
});
if timed_out {
let forwarded = user_trace_messages(&trace, "client.budget_forwarded proto=sqlite ");
assert!(
forwarded.iter().any(|m| m.contains("base_ms=none")),
"expected budget-derived forwarded trace, got {forwarded:?}"
);
}
}
#[test]
fn cancel_interrupts_in_flight_statement_and_drains() {
init_test_logging();
let (cx, trace) = traced_cx_with_budget(Budget::INFINITE);
let conn = block_on(async {
match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
let mut fut = Box::pin(conn.query_unchecked(&cx, INFINITE_QUERY, &[]));
let first = block_on(futures_lite::future::poll_once(fut.as_mut()));
assert!(
first.is_none(),
"runaway query must not complete on first poll"
);
std::thread::sleep(Duration::from_millis(100));
cx.cancel_fast(crate::types::CancelKind::User);
let drain_started = std::time::Instant::now();
match block_on(fut) {
Outcome::Cancelled(reason) => {
assert_eq!(reason.kind, crate::types::CancelKind::User);
}
other => panic!("expected cancellation, got {other:?}"),
}
let drain_elapsed = drain_started.elapsed();
assert!(
drain_elapsed < Duration::from_secs(5),
"interrupt must end the runaway statement promptly, took {drain_elapsed:?}"
);
let interrupts = user_trace_messages(&trace, "client.wire_cancel proto=sqlite ");
assert!(
interrupts
.iter()
.any(|m| m.contains("outcome=interrupt_sent")),
"expected interrupt_sent trace, got {interrupts:?}"
);
assert!(
interrupts.iter().any(|m| m.contains("drain=")),
"expected drain-resolution trace, got {interrupts:?}"
);
let fresh_cx = create_test_cx();
block_on(async {
match conn.query_unchecked(&fresh_cx, "SELECT 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after drain: {other:?}"),
}
});
}
#[test]
fn sqlite_p5_queued_cancel_does_not_interrupt_connection_owner() {
init_test_logging();
let owner_cx = create_test_cx();
let (queued_cx, queued_trace) = traced_cx_with_budget(Budget::INFINITE);
let conn = block_on(async {
match SqliteConnection::open_in_memory(&owner_cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1);
let mut owner = Box::pin(
conn.run_connection_op(&owner_cx, "queue_owner", move |raw| {
run_signalled_infinite_query(raw, started_tx)
}),
);
assert!(
block_on(futures_lite::future::poll_once(owner.as_mut())).is_none(),
"owner must remain in its runaway statement"
);
started_rx
.recv_timeout(Duration::from_secs(5))
.expect("owner statement must start");
let executions = Arc::new(AtomicUsize::new(0));
let queued_executions = Arc::clone(&executions);
let mut queued = Box::pin(conn.run_connection_op(&queued_cx, "queued", move |_| {
queued_executions.fetch_add(1, Ordering::AcqRel);
Ok(())
}));
assert!(
block_on(futures_lite::future::poll_once(queued.as_mut())).is_none(),
"second operation must queue behind the owner"
);
queued_cx.cancel_fast(crate::types::CancelKind::User);
assert!(
block_on(futures_lite::future::poll_once(queued.as_mut())).is_none(),
"queued cancellation must drain until the worker acknowledges"
);
assert!(
block_on(futures_lite::future::poll_once(owner.as_mut())).is_none(),
"cancelling the queued operation must not interrupt the owner"
);
conn.interrupt();
match block_on(owner) {
Outcome::Err(err) if sqlite_error_is_interrupt(&err) => {}
other => panic!("explicit interrupt must stop the owner: {other:?}"),
}
match block_on(queued) {
Outcome::Cancelled(reason) => {
assert_eq!(reason.kind, crate::types::CancelKind::User);
}
other => panic!("queued operation must resolve as cancelled: {other:?}"),
}
assert_eq!(
executions.load(Ordering::Acquire),
0,
"cancelled queued work must never execute"
);
let messages =
user_trace_messages(&queued_trace, "client.wire_cancel proto=sqlite outcome=");
assert!(
messages
.iter()
.any(|message| message.contains("outcome=skipped")
&& message.contains("reason=queued")),
"queued cancellation must record why no interrupt was sent: {messages:?}"
);
}
#[test]
fn sqlite_p5_queued_stream_cancel_does_not_interrupt_connection_owner() {
init_test_logging();
let owner_cx = create_test_cx();
let (stream_cx, stream_trace) = traced_cx_with_budget(Budget::INFINITE);
let mut conn = block_on(async {
match SqliteConnection::open_in_memory(&owner_cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
let owner_interrupt = Arc::clone(&conn.interrupt);
let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1);
let (finished_tx, finished_rx) = std::sync::mpsc::sync_channel(1);
let mut owner =
Box::pin(
conn.run_connection_op(&owner_cx, "stream_queue_owner", move |raw| {
let result = run_signalled_infinite_query(raw, started_tx);
let interrupted = result.as_ref().is_err_and(sqlite_error_is_interrupt);
let _ = finished_tx.send(interrupted);
result
}),
);
assert!(
block_on(futures_lite::future::poll_once(owner.as_mut())).is_none(),
"owner must remain in its runaway statement"
);
started_rx
.recv_timeout(Duration::from_secs(5))
.expect("owner statement must be executing before the stream queues");
drop(owner);
let mut stream = block_on(async {
match conn
.query_stream_unchecked(&stream_cx, "SELECT 1", &[])
.await
{
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
}
});
stream_cx.cancel_fast(crate::types::CancelKind::User);
let mut next = Box::pin(stream.next(&stream_cx));
let first_poll = block_on(futures_lite::future::poll_once(next.as_mut()));
assert!(
matches!(
finished_rx.try_recv(),
Err(std::sync::mpsc::TryRecvError::Empty)
),
"queued stream cancellation must not interrupt the connection owner"
);
owner_interrupt.interrupt();
assert!(
finished_rx
.recv_timeout(Duration::from_secs(5))
.expect("explicit interrupt must finish the owner"),
"owner must finish because of the explicit interrupt"
);
let outcome = match first_poll {
Some(outcome) => {
drop(next);
outcome
}
None => block_on(next),
};
match outcome {
Outcome::Cancelled(reason) => {
assert_eq!(reason.kind, crate::types::CancelKind::User);
}
other => panic!("queued row stream must resolve as cancelled: {other:?}"),
}
drop(stream);
let messages = user_trace_messages(
&stream_trace,
"client.wire_cancel proto=sqlite outcome=skipped op=row_stream",
);
assert!(
messages
.iter()
.any(|message| message.contains("reason=queued")),
"queued stream cancellation must explain why it skipped interrupt: {messages:?}"
);
let fresh_cx = create_test_cx();
block_on(async {
match conn.query_unchecked(&fresh_cx, "SELECT 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after queued stream cancellation: {other:?}"),
}
});
}
#[test]
fn sqlite_p5_reserve_race_cancellation_does_not_execute_operation() {
let setup_cx = create_test_cx();
let cancelled_cx = create_test_cx();
let conn = block_on(async {
match SqliteConnection::open_in_memory(&setup_cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
cancelled_cx.cancel_fast(crate::types::CancelKind::User);
let executions = Arc::new(AtomicUsize::new(0));
let worker_executions = Arc::clone(&executions);
match block_on(
conn.run_connection_op(&cancelled_cx, "reserve_race", move |_| {
worker_executions.fetch_add(1, Ordering::AcqRel);
Ok(())
}),
) {
Outcome::Cancelled(reason) => {
assert_eq!(reason.kind, crate::types::CancelKind::User);
}
other => panic!("pre-start reserve race must cancel: {other:?}"),
}
assert_eq!(executions.load(Ordering::Acquire), 0);
}
#[test]
fn sqlite_p5_committed_result_wins_finishing_cancellation() {
init_test_logging();
let (cx, trace) = traced_cx_with_budget(Budget::INFINITE);
let conn = block_on(async {
match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
block_on(async {
match conn
.execute_batch(&cx, "CREATE TABLE finishing (value INTEGER NOT NULL);")
.await
{
Outcome::Ok(()) => {}
other => panic!("schema setup failed: {other:?}"),
}
});
let (committed_tx, committed_rx) = std::sync::mpsc::sync_channel(1);
let (release_tx, release_rx) = std::sync::mpsc::sync_channel(1);
let mut operation = Box::pin(conn.run_connection_op(&cx, "finishing", move |raw| {
let affected = raw
.execute("INSERT INTO finishing(value) VALUES (7)", [])
.map_err(|error| SqliteError::Sqlite(error.to_string()))?;
committed_tx
.send(())
.expect("commit observer must remain live");
release_rx
.recv()
.expect("test must release result publication");
Ok(affected)
}));
assert!(
block_on(futures_lite::future::poll_once(operation.as_mut())).is_none(),
"operation must park before publishing its committed result"
);
committed_rx
.recv_timeout(Duration::from_secs(5))
.expect("SQLite write must commit before cancellation");
cx.cancel_fast(crate::types::CancelKind::User);
assert!(
block_on(futures_lite::future::poll_once(operation.as_mut())).is_none(),
"cancel drain must wait for the committed result publication"
);
release_tx
.send(())
.expect("blocking operation must still be waiting");
match block_on(operation) {
Outcome::Ok(1) => {}
other => panic!("committed completion must win cancellation: {other:?}"),
}
let fresh_cx = create_test_cx();
block_on(async {
match conn
.query_unchecked(&fresh_cx, "SELECT COUNT(*) AS count FROM finishing", &[])
.await
{
Outcome::Ok(rows) => assert_eq!(
rows[0]
.get_i64("count")
.expect("count column must remain readable"),
1
),
other => panic!("connection unusable after finishing race: {other:?}"),
}
});
let messages = user_trace_messages(&trace, "client.wire_cancel proto=sqlite ");
assert!(
messages
.iter()
.any(|message| message.contains("completion=won")),
"finishing race must record terminal-completion precedence: {messages:?}"
);
}
#[test]
fn sqlite_p5_explicit_interrupt_stops_statement_and_preserves_connection() {
let cx = create_test_cx();
let conn = block_on(async {
match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1);
let mut operation = Box::pin(conn.run_connection_op(
&cx,
"explicit_interrupt",
move |raw| run_signalled_infinite_query(raw, started_tx),
));
assert!(
block_on(futures_lite::future::poll_once(operation.as_mut())).is_none(),
"runaway statement must be in flight"
);
started_rx
.recv_timeout(Duration::from_secs(5))
.expect("statement must start before explicit interrupt");
conn.interrupt();
match block_on(operation) {
Outcome::Err(err) if sqlite_error_is_interrupt(&err) => {}
other => panic!("explicit interrupt must surface SQLite interruption: {other:?}"),
}
block_on(async {
match conn.query_unchecked(&cx, "SELECT 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after explicit interrupt: {other:?}"),
}
});
}
#[test]
fn row_stream_cancel_interrupts_and_drains() {
init_test_logging();
let (cx, trace) = traced_cx_with_budget(Budget::INFINITE);
let mut conn = block_on(async {
match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
let mut stream = block_on(async {
match conn.query_stream_unchecked(&cx, INFINITE_QUERY, &[]).await {
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed: {other:?}"),
}
});
let mut fut = Box::pin(stream.next(&cx));
let first = block_on(futures_lite::future::poll_once(fut.as_mut()));
assert!(first.is_none(), "runaway stream must not yield a first row");
std::thread::sleep(Duration::from_millis(100));
cx.cancel_fast(crate::types::CancelKind::User);
let drain_started = std::time::Instant::now();
match block_on(fut) {
Outcome::Cancelled(reason) => {
assert_eq!(reason.kind, crate::types::CancelKind::User);
}
other => panic!("expected cancellation, got {other:?}"),
}
assert!(
drain_started.elapsed() < Duration::from_secs(5),
"interrupt must end the runaway stream promptly"
);
let interrupts = user_trace_messages(&trace, "client.wire_cancel proto=sqlite ");
assert!(
interrupts
.iter()
.any(|m| m.contains("outcome=interrupt_sent") && m.contains("op=row_stream")),
"expected stream interrupt trace, got {interrupts:?}"
);
drop(stream);
let fresh_cx = create_test_cx();
block_on(async {
match conn.query_unchecked(&fresh_cx, "SELECT 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after stream drain: {other:?}"),
}
});
}
#[test]
fn row_stream_statement_timeout_aborts_runaway_query() {
init_test_logging();
let (cx, trace) = traced_cx_with_budget(Budget::INFINITE);
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
conn.set_statement_timeout_override(Some(Duration::from_millis(50)));
let mut stream = match conn.query_stream_unchecked(&cx, INFINITE_QUERY, &[]).await {
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed: {other:?}"),
};
match stream.next(&cx).await {
Outcome::Err(SqliteError::StatementTimeout { limit }) => {
assert_eq!(limit, Duration::from_millis(50));
}
other => panic!("expected StatementTimeout from stream, got {other:?}"),
}
});
let forwarded = user_trace_messages(&trace, "client.budget_forwarded proto=sqlite ");
assert!(
forwarded
.iter()
.any(|m| m.contains("op=row_stream") && m.contains("statement_timeout_ms=50")),
"expected stream forwarded-budget trace, got {forwarded:?}"
);
}
#[test]
fn dropped_row_stream_interrupts_runaway_statement() {
init_test_logging();
let cx = create_test_cx();
let mut conn = block_on(async {
match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
}
});
let stream = block_on(async {
match conn.query_stream_unchecked(&cx, INFINITE_QUERY, &[]).await {
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed: {other:?}"),
}
});
std::thread::sleep(Duration::from_millis(100));
drop(stream);
let started = std::time::Instant::now();
block_on(async {
match conn.query_unchecked(&cx, "SELECT 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after stream drop: {other:?}"),
}
});
assert!(
started.elapsed() < Duration::from_secs(5),
"dropped stream must free the connection promptly, took {:?}",
started.elapsed()
);
}
#[test]
fn test_sqlite_value_display() {
assert_eq!(SqliteValue::Null.to_string(), "NULL");
assert_eq!(SqliteValue::Integer(42).to_string(), "42");
assert_eq!(SqliteValue::Real(3.5).to_string(), "3.5");
assert_eq!(SqliteValue::Text("hello".to_string()).to_string(), "hello");
assert_eq!(
SqliteValue::Blob(vec![1, 2, 3]).to_string(),
"<blob 3 bytes>"
);
}
#[test]
fn test_sqlite_value_accessors() {
assert!(SqliteValue::Null.is_null());
assert!(!SqliteValue::Integer(42).is_null());
assert_eq!(SqliteValue::Integer(42).as_integer(), Some(42));
assert_eq!(SqliteValue::Text("hi".to_string()).as_integer(), None);
assert_eq!(SqliteValue::Real(3.5).as_real(), Some(3.5));
assert_eq!(SqliteValue::Integer(42).as_real(), Some(42.0));
assert_eq!(SqliteValue::Real(3.5).as_real_strict(), Some(3.5));
assert_eq!(SqliteValue::Integer(42).as_real_strict(), None);
assert_eq!(
SqliteValue::Text("hello".to_string()).as_text(),
Some("hello")
);
assert_eq!(SqliteValue::Integer(42).as_text(), None);
assert_eq!(
SqliteValue::Blob(vec![1, 2, 3]).as_blob(),
Some(&[1, 2, 3][..])
);
}
#[test]
fn test_sqlite_row_accessors() {
let mut columns = BTreeMap::new();
columns.insert("id".to_string(), 0);
columns.insert("name".to_string(), 1);
let columns = Arc::new(columns);
let values = vec![
SqliteValue::Integer(1),
SqliteValue::Text("Alice".to_string()),
];
let ordered_columns: Arc<[String]> = vec!["id".to_string(), "name".to_string()].into();
let row = SqliteRow::new(columns, ordered_columns, values);
assert_eq!(row.len(), 2);
assert!(!row.is_empty());
assert_eq!(row.get_i64("id").unwrap(), 1);
assert_eq!(row.get_str("name").unwrap(), "Alice");
assert!(row.get("missing").is_err());
}
#[test]
fn sqlite_error_display_sqlite() {
let err = SqliteError::Sqlite("connection refused".into());
assert_eq!(err.to_string(), "SQLite error: connection refused");
}
#[test]
fn sqlite_error_display_cancelled() {
let err = SqliteError::Cancelled(CancelReason::user("timeout"));
let msg = err.to_string();
assert!(msg.starts_with("SQLite operation cancelled:"), "{msg}");
}
#[test]
fn sqlite_error_display_connection_closed() {
assert_eq!(
SqliteError::ConnectionClosed.to_string(),
"SQLite connection is closed"
);
}
#[test]
fn sqlite_error_display_column_not_found() {
let err = SqliteError::ColumnNotFound("missing_col".into());
assert_eq!(err.to_string(), "Column not found: missing_col");
}
#[test]
fn sqlite_error_display_type_mismatch() {
let err = SqliteError::TypeMismatch {
column: "age".into(),
expected: "integer",
actual: "Text(\"hello\")".into(),
};
assert_eq!(
err.to_string(),
"Type mismatch for column age: expected integer, got Text(\"hello\")"
);
}
#[test]
fn sqlite_error_display_io() {
let io_err = std::io::Error::new(std::io::ErrorKind::NotFound, "file not found");
let err = SqliteError::Io(io_err);
assert!(err.to_string().starts_with("SQLite I/O error:"), "{err}");
}
#[test]
fn sqlite_error_display_transaction_finished() {
assert_eq!(
SqliteError::TransactionFinished.to_string(),
"Transaction already finished"
);
}
#[test]
fn sqlite_error_display_lock_poisoned() {
assert_eq!(
SqliteError::LockPoisoned.to_string(),
"SQLite connection lock poisoned"
);
}
#[test]
fn sqlite_error_display_unsafe_sql() {
let err = SqliteError::UnsafeSql("PRAGMA statements require *_unchecked".into());
assert_eq!(
err.to_string(),
"Unsafe SQLite control SQL on SQLite binding surface: PRAGMA statements require *_unchecked"
);
}
#[test]
fn sqlite_error_display_unsafe_path() {
let err = SqliteError::UnsafePath("resolved into /etc".into());
assert_eq!(
err.to_string(),
"Unsafe SQLite database path: resolved into /etc"
);
}
#[test]
fn sqlite_error_display_invalid_text_encoding() {
let invalid_utf8 = vec![0x80_u8];
let err = SqliteError::InvalidTextEncoding {
column: "payload".into(),
source: std::str::from_utf8(&invalid_utf8).unwrap_err(),
};
assert!(
err.to_string()
.starts_with("SQLite text column payload contained invalid UTF-8:")
);
}
#[test]
fn sqlite_error_source_io_returns_some() {
use std::error::Error;
let io_err = std::io::Error::other("disk failure");
let err = SqliteError::Io(io_err);
assert!(err.source().is_some());
}
#[test]
fn sqlite_error_source_non_io_returns_none() {
use std::error::Error;
assert!(SqliteError::ConnectionClosed.source().is_none());
assert!(SqliteError::Sqlite("oops".into()).source().is_none());
assert!(SqliteError::LockPoisoned.source().is_none());
assert!(SqliteError::TransactionFinished.source().is_none());
assert!(SqliteError::UnsafeSql("oops".into()).source().is_none());
assert!(SqliteError::ColumnNotFound("x".into()).source().is_none());
}
#[test]
fn sqlite_error_source_invalid_text_encoding_returns_some() {
use std::error::Error;
let invalid_utf8 = vec![0x80_u8];
let err = SqliteError::InvalidTextEncoding {
column: "payload".into(),
source: std::str::from_utf8(&invalid_utf8).unwrap_err(),
};
assert!(err.source().is_some());
}
#[test]
fn checked_sql_surface_rejects_transaction_control_keywords() {
for sql in [
"BEGIN IMMEDIATE",
" -- comment\nROLLBACK",
"/* comment */ SAVEPOINT sp1",
"ATTACH 'tenant.db' AS tenant",
] {
let err = ensure_checked_sql_surface(sql, CheckedSqlCardinality::Batch).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(_)),
"expected unsafe SQL rejection for {sql:?}, got {err:?}"
);
}
}
#[test]
fn checked_sql_surface_rejects_pragma_keywords() {
for sql in [
"PRAGMA read_uncommitted = 1",
" /* comment */ PRAGMA foreign_keys = OFF",
] {
let err = ensure_checked_sql_surface(sql, CheckedSqlCardinality::Batch).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(_)),
"expected unsafe SQL rejection for {sql:?}, got {err:?}"
);
}
}
#[test]
fn unchecked_sql_surface_rejects_attach_detach_keywords() {
for sql in ["ATTACH 'tenant.db' AS tenant", "DETACH tenant"] {
let err = ensure_unchecked_sql_surface(sql).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(_)),
"expected unsafe SQL rejection for {sql:?}, got {err:?}"
);
}
}
#[test]
fn unchecked_sql_surface_allows_pragma_and_transaction_control() {
for sql in ["PRAGMA journal_mode", "BEGIN IMMEDIATE", "ROLLBACK"] {
ensure_unchecked_sql_surface(sql)
.unwrap_or_else(|err| panic!("unchecked surface should allow {sql:?}: {err:?}"));
}
}
#[test]
fn unchecked_sql_surface_preserves_large_trusted_migration_compatibility() {
let oversized = format!("SELECT '{}';", "x".repeat(MAX_CHECKED_SQL_BYTES));
ensure_unchecked_sql_surface(&oversized)
.expect("the explicit unchecked surface must not inherit checked parser size limits");
let oversized_attach = format!(
"SELECT '{}'; VACUUM; ATTACH ':memory:' AS bypass",
"x".repeat(MAX_CHECKED_SQL_BYTES)
);
let err = ensure_unchecked_sql_surface(&oversized_attach).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(_)),
"oversized unchecked batches must not bypass the permanent ATTACH ban: {err:?}"
);
}
#[test]
fn validate_sqlite_open_path_rejects_tilde_prefixes() {
for raw in ["~/tenant.db", "~alice/tenant.db"] {
let err = validate_sqlite_open_path(Path::new(raw)).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafePath(ref msg) if msg.contains("tilde-prefixed")),
"expected tilde rejection for {raw:?}, got {err:?}"
);
}
}
#[test]
fn validate_sqlite_open_path_rejects_restricted_system_directory() {
let err = validate_sqlite_open_path(Path::new("/etc/asupersync-test.sqlite")).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafePath(ref msg) if msg.contains("/etc")),
"expected /etc rejection, got {err:?}"
);
}
#[test]
fn validate_sqlite_open_path_rejects_parent_directory_traversal() {
for raw in ["../tenant.db", "nested/../../tenant.db"] {
let err = validate_sqlite_open_path(Path::new(raw)).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafePath(ref msg) if msg.contains("parent-directory traversal")),
"expected traversal rejection for {raw:?}, got {err:?}"
);
}
}
#[test]
fn checked_sql_surface_allows_regular_dml() {
for sql in [
"SELECT * FROM users",
"INSERT INTO users(name) VALUES ('alice')",
"WITH cte AS (SELECT 1) SELECT * FROM cte",
] {
ensure_checked_sql_surface(sql, CheckedSqlCardinality::ExactlyOne)
.unwrap_or_else(|err| panic!("checked surface should allow {sql:?}: {err:?}"));
}
}
#[test]
fn checked_sql_surface_allows_create_trigger_ddl() {
let sql = "
CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT);
CREATE TRIGGER t_audit
AFTER INSERT ON t
BEGIN
INSERT INTO t(name) VALUES ('copied;still literal');
END;
";
ensure_checked_sql_surface(sql, CheckedSqlCardinality::Batch)
.unwrap_or_else(|err| panic!("checked surface should allow trigger DDL: {err:?}"));
}
#[test]
fn checked_sql_surface_rejects_top_level_end_transaction_control() {
let err = ensure_checked_sql_surface("END", CheckedSqlCardinality::ExactlyOne).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(_)),
"expected unsafe SQL rejection for END, got {err:?}"
);
}
#[test]
fn checked_sql_surface_rejects_vacuum_and_vacuum_into() {
for sql in [
"VACUUM",
"VACUUM main",
"VACUUM INTO '/tmp/asupersync-copy.sqlite'",
"/* audited? */ VACUUM\tINTO '/tmp/asupersync-copy.sqlite'",
] {
let err =
ensure_checked_sql_surface(sql, CheckedSqlCardinality::ExactlyOne).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(ref msg) if msg.starts_with("VACUUM requires")),
"expected VACUUM rejection for {sql:?}, got {err:?}"
);
}
ensure_unchecked_sql_surface("VACUUM")
.expect("the explicit unchecked surface retains ordinary VACUUM compatibility");
}
#[test]
fn checked_sql_surface_enforces_statement_count_and_parser_limits() {
let err =
ensure_checked_sql_surface("SELECT 1; SELECT 2", CheckedSqlCardinality::ExactlyOne)
.unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(ref msg) if msg.contains("exactly one")),
"single-statement APIs must reject multiple statements: {err:?}"
);
ensure_checked_sql_surface("SELECT 1; SELECT 2", CheckedSqlCardinality::Batch)
.expect("checked batch APIs retain multiple-statement support");
let malformed =
ensure_checked_sql_surface("SELECT FROM", CheckedSqlCardinality::ExactlyOne)
.unwrap_err();
assert!(
matches!(malformed, SqliteError::UnsafeSql(ref msg) if msg.contains("bounded policy parser")),
"parser divergence must fail closed: {malformed:?}"
);
let oversized = format!("SELECT '{}';", "x".repeat(MAX_CHECKED_SQL_BYTES));
let err =
ensure_checked_sql_surface(&oversized, CheckedSqlCardinality::ExactlyOne).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(ref msg) if msg.contains("resource limits")),
"oversized checked SQL must fail before parsing: {err:?}"
);
let nested = format!(
"SELECT {}1{}",
"(".repeat(MAX_CHECKED_SQL_RECURSION + 8),
")".repeat(MAX_CHECKED_SQL_RECURSION + 8)
);
let err =
ensure_checked_sql_surface(&nested, CheckedSqlCardinality::ExactlyOne).unwrap_err();
assert!(
matches!(err, SqliteError::UnsafeSql(ref msg) if msg.contains("bounded policy parser")),
"excessive parser recursion must fail closed: {err:?}"
);
}
#[test]
fn checked_sql_surface_does_not_treat_comments_or_literals_as_control_sql() {
for sql in [
"SELECT 'VACUUM INTO /tmp/copy.sqlite'",
"SELECT 'ATTACH tenant.sqlite'",
"-- PRAGMA foreign_keys=OFF\nSELECT 1",
"/* ROLLBACK; ATTACH x */ SELECT 'Δatabase'",
] {
ensure_checked_sql_surface(sql, CheckedSqlCardinality::ExactlyOne).unwrap_or_else(
|err| panic!("quoted/commented control words are data for {sql:?}: {err:?}"),
);
}
}
#[test]
fn fallback_keyword_boundaries_cover_sqlite_whitespace_and_punctuation() {
for sql in [
"PRAGMA(main.table_info)",
"ATTACH\t'db.sqlite' AS tenant",
"DETACH\ntenant",
"VACUUM\tINTO 'copy.sqlite'",
"RELEASE\tSAVEPOINT sp",
] {
assert!(
check_sql_keywords_fallback(sql).is_some(),
"fallback must classify control statement {sql:?}"
);
}
assert_eq!(check_sql_keywords_fallback("BEGINNING SELECT 1"), None);
assert_eq!(check_sql_keywords_fallback("VACUUMED SELECT 1"), None);
}
#[test]
fn checked_sql_policy_bounded_adversarial_fuzz_is_panic_free() {
let mut state = 0x5eed_5eed_cafe_f00d_u64;
for case_index in 0..4096_u64 {
state = state
.wrapping_mul(6_364_136_223_846_793_005)
.wrapping_add(1_442_695_040_888_963_407);
let padding = " ".repeat((state as usize) & 15);
let marker = format!("{state:016x}");
let (sql, should_allow) = match case_index % 6 {
0 => (
format!("/*{marker}*/{padding}PRAGMA foreign_keys=OFF"),
false,
),
1 => (
format!("--{marker}\n{padding}ATTACH ':memory:' AS escaped"),
false,
),
2 => (
format!("SELECT load_extension{padding}('/tmp/{marker}.so')"),
false,
),
3 => (format!("SELECT 'Δ-{marker}-\u{200b}'"), true),
4 => (format!("SELECT ({padding}{marker}"), false),
_ => (
format!("SELECT 1; /*{marker}*/ VACUUM INTO '/tmp/{marker}.db'"),
false,
),
};
let result = std::panic::catch_unwind(|| validate_checked_sql_statement(&sql))
.unwrap_or_else(|_| {
panic!("checked policy panicked for case {case_index}: {sql:?}")
});
assert_eq!(
result.is_ok(),
should_allow,
"unexpected policy result for case {case_index}: {sql:?}: {result:?}"
);
}
}
#[test]
fn every_checked_public_entry_point_applies_the_same_fail_closed_policy() {
let cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
assert!(matches!(
conn.execute(&cx, "ATTACH ':memory:' AS blocked", &[]).await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
assert!(matches!(
conn.execute_batch(&cx, "SELECT 1; PRAGMA foreign_keys=OFF")
.await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
assert!(matches!(
conn.query(&cx, "SELECT load_extension('/tmp/blocked.so')", &[])
.await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
assert!(matches!(
conn.query_row(&cx, "VACUUM INTO '/tmp/blocked.db'", &[])
.await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
assert!(matches!(
conn.query_stream(&cx, "BEGIN IMMEDIATE", &[]).await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
match conn.query_row(&cx, "SELECT 1 AS value", &[]).await {
Outcome::Ok(Some(row)) => {
assert!(matches!(row.get("value"), Ok(SqliteValue::Integer(1))));
}
other => panic!("connection was not reusable after policy rejection: {other:?}"),
}
let control_text = "ATTACH PRAGMA VACUUM load_extension(";
match conn
.query_row(
&cx,
"SELECT ?1 AS value",
&[SqliteValue::Text(control_text.to_owned())],
)
.await
{
Outcome::Ok(Some(row)) => match row.get("value") {
Ok(SqliteValue::Text(value)) => assert_eq!(value, control_text),
other => panic!("bound value had the wrong shape: {other:?}"),
},
other => panic!("bound control-like data was not preserved: {other:?}"),
}
let transaction = match conn.begin(&cx).await {
Outcome::Ok(transaction) => transaction,
_ => panic!("begin failed after checked-policy rejections"),
};
assert!(matches!(
transaction
.execute(&cx, "PRAGMA writable_schema=ON", &[])
.await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
assert!(matches!(
transaction
.query(&cx, "SELECT load_extension('/tmp/blocked.so')", &[])
.await,
Outcome::Err(SqliteError::UnsafeSql(_))
));
assert!(matches!(transaction.rollback(&cx).await, Outcome::Ok(())));
});
}
#[test]
fn sqlite_error_from_io_error() {
let io_err = std::io::Error::new(std::io::ErrorKind::PermissionDenied, "denied");
let err: SqliteError = io_err.into();
assert!(matches!(err, SqliteError::Io(_)));
}
#[test]
fn sqlite_value_partial_eq() {
assert_eq!(SqliteValue::Null, SqliteValue::Null);
assert_eq!(SqliteValue::Integer(10), SqliteValue::Integer(10));
assert_ne!(SqliteValue::Integer(10), SqliteValue::Integer(20));
assert_eq!(SqliteValue::Real(1.5), SqliteValue::Real(1.5));
assert_eq!(SqliteValue::Text("a".into()), SqliteValue::Text("a".into()));
assert_ne!(SqliteValue::Text("a".into()), SqliteValue::Text("b".into()));
assert_eq!(SqliteValue::Blob(vec![1, 2]), SqliteValue::Blob(vec![1, 2]));
assert_ne!(SqliteValue::Null, SqliteValue::Integer(0));
}
#[test]
fn sqlite_value_as_real_returns_none_for_text() {
assert_eq!(SqliteValue::Text("nope".into()).as_real(), None);
}
#[test]
fn sqlite_value_as_real_returns_none_for_blob() {
assert_eq!(SqliteValue::Blob(vec![1]).as_real(), None);
}
#[test]
fn sqlite_value_as_real_returns_none_for_null() {
assert_eq!(SqliteValue::Null.as_real(), None);
}
#[test]
fn sqlite_value_as_integer_returns_none_for_real() {
assert_eq!(SqliteValue::Real(3.5).as_integer(), None);
}
#[test]
fn sqlite_value_as_text_returns_none_for_blob() {
assert_eq!(SqliteValue::Blob(vec![0]).as_text(), None);
}
#[test]
fn sqlite_value_as_blob_returns_none_for_text() {
assert_eq!(SqliteValue::Text("x".into()).as_blob(), None);
}
#[test]
fn sqlite_value_as_blob_returns_none_for_null() {
assert_eq!(SqliteValue::Null.as_blob(), None);
}
#[test]
fn sqlite_value_display_empty_blob() {
assert_eq!(SqliteValue::Blob(vec![]).to_string(), "<blob 0 bytes>");
}
#[test]
fn sqlite_value_display_negative_integer() {
assert_eq!(SqliteValue::Integer(-99).to_string(), "-99");
}
fn make_test_sqlite_row(names: &[&str], values: Vec<SqliteValue>) -> SqliteRow {
let mut columns = BTreeMap::new();
for (i, name) in names.iter().enumerate() {
columns.insert(name.to_string(), i);
}
let ordered_columns = names
.iter()
.map(|name| (*name).to_string())
.collect::<Vec<_>>()
.into();
SqliteRow::new(Arc::new(columns), ordered_columns, values)
}
#[test]
fn sqlite_row_get_idx_valid() {
let row = make_test_sqlite_row(
&["a", "b"],
vec![SqliteValue::Integer(1), SqliteValue::Text("two".into())],
);
assert_eq!(row.get_idx(0).unwrap(), &SqliteValue::Integer(1));
assert_eq!(row.get_idx(1).unwrap(), &SqliteValue::Text("two".into()));
}
#[test]
fn sqlite_row_get_idx_out_of_bounds() {
let row = make_test_sqlite_row(&["a"], vec![SqliteValue::Null]);
assert!(row.get_idx(5).is_err());
}
#[test]
fn sqlite_row_get_f64_success() {
let row = make_test_sqlite_row(&["val"], vec![SqliteValue::Real(3.5)]);
assert!((row.get_f64("val").unwrap() - 3.5).abs() < f64::EPSILON);
}
#[test]
fn sqlite_row_get_f64_widens_from_integer() {
let row = make_test_sqlite_row(&["val"], vec![SqliteValue::Integer(7)]);
assert!((row.get_f64("val").unwrap() - 7.0).abs() < f64::EPSILON);
assert!(matches!(
row.get_f64_strict("val"),
Err(SqliteError::TypeMismatch { .. })
));
}
#[test]
fn sqlite_row_get_f64_strict_accepts_real() {
let row = make_test_sqlite_row(&["val"], vec![SqliteValue::Real(3.5)]);
assert_eq!(row.get_f64_strict("val").unwrap(), 3.5);
}
#[test]
fn sqlite_row_get_f64_type_mismatch() {
let row = make_test_sqlite_row(&["name"], vec![SqliteValue::Text("alice".into())]);
let err = row.get_f64("name").unwrap_err();
assert!(matches!(err, SqliteError::TypeMismatch { .. }));
}
#[test]
fn sqlite_row_get_blob_success() {
let row = make_test_sqlite_row(&["data"], vec![SqliteValue::Blob(vec![0xDE, 0xAD])]);
assert_eq!(row.get_blob("data").unwrap(), &[0xDE, 0xAD]);
}
#[test]
fn sqlite_row_get_blob_type_mismatch() {
let row = make_test_sqlite_row(&["num"], vec![SqliteValue::Integer(42)]);
let err = row.get_blob("num").unwrap_err();
assert!(matches!(err, SqliteError::TypeMismatch { .. }));
}
#[test]
fn sqlite_row_get_i64_type_mismatch() {
let row = make_test_sqlite_row(&["name"], vec![SqliteValue::Text("not_a_number".into())]);
let err = row.get_i64("name").unwrap_err();
assert!(matches!(err, SqliteError::TypeMismatch { .. }));
}
#[test]
fn sqlite_row_get_str_type_mismatch() {
let row = make_test_sqlite_row(&["id"], vec![SqliteValue::Integer(1)]);
let err = row.get_str("id").unwrap_err();
assert!(matches!(err, SqliteError::TypeMismatch { .. }));
}
#[test]
fn sqlite_row_column_names() {
let row = make_test_sqlite_row(
&["alpha", "beta", "gamma"],
vec![SqliteValue::Null, SqliteValue::Null, SqliteValue::Null],
);
let names: Vec<&str> = row.column_names().collect();
assert_eq!(names, vec!["alpha", "beta", "gamma"]);
}
#[test]
fn sqlite_row_debug_preserves_v043_shape() {
let row = make_test_sqlite_row(&["id"], vec![SqliteValue::Integer(7)]);
assert_eq!(
format!("{row:?}"),
"SqliteRow { columns: {\"id\": 0}, values: [Integer(7)] }"
);
}
fn assert_duplicate_sqlite_row_metadata(row: &SqliteRow) {
let ordered: Vec<&str> = row.column_names_in_order().collect();
assert_eq!(ordered, vec!["dup", "Beta", "dup", "alpha"]);
assert_eq!(row.column_name(0), Some("dup"));
assert_eq!(row.column_name(3), Some("alpha"));
assert_eq!(row.column_name(4), None);
assert_eq!(row.column_index("dup"), Some(0));
assert_eq!(row.column_index("DUP"), Some(0));
assert_eq!(row.column_index("beta"), Some(1));
assert_eq!(row.column_index("missing"), None);
assert_eq!(row.get_idx(0).unwrap(), &SqliteValue::Integer(10));
assert_eq!(row.get_idx(2).unwrap(), &SqliteValue::Integer(30));
assert_eq!(row.get_i64("dup").unwrap(), 30);
assert!(matches!(
row.get("DUP"),
Err(SqliteError::ColumnNotFound(name)) if name == "DUP"
));
assert_eq!(
row.column_names().collect::<Vec<_>>(),
vec!["Beta", "alpha", "dup"]
);
assert_eq!(
format!("{row:?}"),
"SqliteRow { columns: {\"Beta\": 1, \"alpha\": 3, \"dup\": 2}, values: [Integer(10), Integer(20), Integer(30), Integer(40)] }"
);
}
#[test]
fn sqlite_row_ordered_metadata_preserves_duplicates_across_query_surfaces() {
const DUPLICATE_COLUMNS: &str = "SELECT 10 AS dup, 20 AS Beta, 30 AS dup, 40 AS alpha";
let cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
let rows = match conn.query(&cx, DUPLICATE_COLUMNS, &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("query failed: {other:?}"),
};
assert_eq!(rows.len(), 1);
assert_duplicate_sqlite_row_metadata(&rows[0]);
let row = match conn.query_row(&cx, DUPLICATE_COLUMNS, &[]).await {
Outcome::Ok(Some(row)) => row,
other => panic!("query_row failed: {other:?}"),
};
assert_duplicate_sqlite_row_metadata(&row);
let mut stream = match conn.query_stream(&cx, DUPLICATE_COLUMNS, &[]).await {
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
};
let row = match stream.next(&cx).await {
Outcome::Ok(Some(row)) => row,
other => panic!("query_stream first row failed: {other:?}"),
};
assert_duplicate_sqlite_row_metadata(&row);
assert!(matches!(stream.next(&cx).await, Outcome::Ok(None)));
});
}
fn assert_sqlite_value_boundary_row(row: &SqliteRow) {
const ABOVE_EXACT_BINARY64_INTEGER: i64 = (1_i64 << 53) + 1;
assert_eq!(row.get_i64("int_min").unwrap(), i64::MIN);
assert_eq!(row.get_i64("int_max").unwrap(), i64::MAX);
assert_eq!(
row.get_i64("above_exact").unwrap(),
ABOVE_EXACT_BINARY64_INTEGER
);
assert_eq!(row.get_f64("above_exact").unwrap(), 9_007_199_254_740_992.0);
assert!(matches!(
row.get_f64_strict("above_exact"),
Err(SqliteError::TypeMismatch { .. })
));
assert_eq!(
row.get_f64_strict("negative_zero").unwrap().to_bits(),
(-0.0_f64).to_bits()
);
assert_eq!(
row.get_f64_strict("positive_infinity").unwrap(),
f64::INFINITY
);
assert_eq!(
row.get_f64_strict("negative_infinity").unwrap(),
f64::NEG_INFINITY
);
assert_eq!(row.get("nan_value").unwrap(), &SqliteValue::Null);
assert_eq!(row.get_str("empty_text").unwrap(), "");
assert_eq!(row.get_str("nul_text").unwrap(), "a\0b");
assert_eq!(row.get_str("unicode_text").unwrap(), "e\u{301}雪");
assert_eq!(row.get_blob("empty_blob").unwrap(), b"");
assert_eq!(row.get_blob("binary_blob").unwrap(), &[0x00, 0x80, 0xff]);
}
#[test]
fn sqlite_value_boundaries_round_trip_across_query_surfaces() {
const VALUE_QUERY: &str = "SELECT ?1 AS int_min, ?2 AS int_max, \
?3 AS above_exact, ?4 AS negative_zero, ?5 AS positive_infinity, \
?6 AS negative_infinity, ?7 AS nan_value, ?8 AS empty_text, \
?9 AS nul_text, ?10 AS unicode_text, ?11 AS empty_blob, \
?12 AS binary_blob";
let cx = create_test_cx();
let (query_row, one_row, streamed_row) = block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
let params = vec![
SqliteValue::Integer(i64::MIN),
SqliteValue::Integer(i64::MAX),
SqliteValue::Integer((1_i64 << 53) + 1),
SqliteValue::Real(-0.0),
SqliteValue::Real(f64::INFINITY),
SqliteValue::Real(f64::NEG_INFINITY),
SqliteValue::Real(f64::NAN),
SqliteValue::Text(String::new()),
SqliteValue::Text("a\0b".to_string()),
SqliteValue::Text("e\u{301}雪".to_string()),
SqliteValue::Blob(Vec::new()),
SqliteValue::Blob(vec![0x00, 0x80, 0xff]),
];
let mut rows = match conn.query(&cx, VALUE_QUERY, ¶ms).await {
Outcome::Ok(rows) => rows,
other => panic!("query failed: {other:?}"),
};
assert_eq!(rows.len(), 1);
let query_row = rows.remove(0);
let one_row = match conn.query_row(&cx, VALUE_QUERY, ¶ms).await {
Outcome::Ok(Some(row)) => row,
other => panic!("query_row failed: {other:?}"),
};
let mut stream = match conn.query_stream(&cx, VALUE_QUERY, ¶ms).await {
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
};
let streamed_row = match stream.next(&cx).await {
Outcome::Ok(Some(row)) => row,
other => panic!("query_stream first row failed: {other:?}"),
};
assert!(matches!(stream.next(&cx).await, Outcome::Ok(None)));
drop(stream);
conn.close().unwrap();
(query_row, one_row, streamed_row)
});
for row in [&query_row, &one_row, &streamed_row] {
assert_sqlite_value_boundary_row(row);
}
}
#[test]
fn sqlite_row_empty() {
let row = make_test_sqlite_row(&[], vec![]);
assert_eq!(row.len(), 0);
assert!(row.is_empty());
assert!(row.get_idx(0).is_err());
assert_eq!(row.column_names().count(), 0);
}
#[test]
fn sqlite_row_get_column_not_found() {
let row = make_test_sqlite_row(&["exists"], vec![SqliteValue::Integer(1)]);
let err = row.get("nope").unwrap_err();
assert!(matches!(err, SqliteError::ColumnNotFound(_)));
}
#[test]
fn test_open_in_memory_exec_query_round_trip() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
match conn
.execute(
&cx,
"INSERT INTO t(name) VALUES (?1)",
&[SqliteValue::Text("alice".to_string())],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("insert failed: {other:?}"),
}
let rows = match conn.query(&cx, "SELECT name FROM t", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("query failed: {other:?}"),
};
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get_str("name").unwrap(), "alice");
});
}
#[test]
fn sqlite_query_stream_yields_many_rows_with_single_row_buffer() {
let cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE streamed (id INTEGER PRIMARY KEY, payload TEXT);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("create streamed table failed: {other:?}"),
}
for id in 0..64 {
let payload = format!("payload-{id:03}-{}", "x".repeat(1024));
match conn
.execute(
&cx,
"INSERT INTO streamed(id, payload) VALUES (?1, ?2)",
&[SqliteValue::Integer(id), SqliteValue::Text(payload)],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("streamed insert {id} failed: {other:?}"),
}
}
let mut stream = match conn
.query_stream(&cx, "SELECT id, payload FROM streamed ORDER BY id", &[])
.await
{
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
};
let mut ids = Vec::new();
while let Outcome::Ok(Some(row)) = stream.next(&cx).await {
ids.push(row.get_i64("id").unwrap());
assert_eq!(
row.get_str("payload").unwrap().len(),
"payload-000-".len().saturating_add(1024)
);
}
let stats = stream.stats();
assert_eq!(ids, (0..64).collect::<Vec<_>>());
assert_eq!(stats.rows_yielded, 64);
assert_eq!(stats.rows_stepped, 64);
assert_eq!(stats.buffered_rows, 0);
assert_eq!(stats.channel_capacity, SQLITE_ROW_STREAM_CHANNEL_CAPACITY);
assert!(
stats.peak_buffered_rows <= SQLITE_ROW_STREAM_CHANNEL_CAPACITY,
"SQLite row stream must not buffer more than one row: {stats:?}"
);
});
}
#[test]
fn sqlite_query_stream_drop_finalizes_statement_and_returns_connection() {
let cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE streamed_drop (id INTEGER PRIMARY KEY);
INSERT INTO streamed_drop(id) VALUES (1), (2), (3), (4);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("create streamed_drop table failed: {other:?}"),
}
let mut stream = match conn
.query_stream(&cx, "SELECT id FROM streamed_drop ORDER BY id", &[])
.await
{
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
};
match stream.next(&cx).await {
Outcome::Ok(Some(row)) => assert_eq!(row.get_i64("id").unwrap(), 1),
other => panic!("first stream row failed: {other:?}"),
}
drop(stream);
let rows = match conn
.query(&cx, "SELECT COUNT(*) AS count FROM streamed_drop", &[])
.await
{
Outcome::Ok(rows) => rows,
other => panic!("connection was not returned after stream drop: {other:?}"),
};
assert_eq!(rows[0].get_i64("count").unwrap(), 4);
});
}
#[test]
fn sqlite_query_stream_surfaces_query_error_on_next() {
let cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
let mut stream = match conn
.query_stream(&cx, "SELECT value FROM missing_table", &[])
.await
{
Outcome::Ok(stream) => stream,
other => panic!("query_stream should defer SQLite prepare errors: {other:?}"),
};
match stream.next(&cx).await {
Outcome::Err(SqliteError::Sqlite(message)) => {
assert!(
message.contains("missing_table") || message.contains("no such table"),
"unexpected SQLite error: {message}"
);
}
other => panic!("missing table should surface through stream next: {other:?}"),
}
});
}
#[test]
fn sqlite_query_stream_cancelled_next_closes_stream_and_connection_recovers() {
let cx = create_test_cx();
let cancel_cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE streamed_cancel (id INTEGER PRIMARY KEY);
INSERT INTO streamed_cancel(id) VALUES (1), (2), (3);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("create streamed_cancel table failed: {other:?}"),
}
let mut stream = match conn
.query_stream(&cx, "SELECT id FROM streamed_cancel ORDER BY id", &[])
.await
{
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
};
cancel_cx.set_cancel_requested(true);
match stream.next(&cancel_cx).await {
Outcome::Cancelled(_) => {}
other => panic!("cancelled stream next should return Cancelled: {other:?}"),
}
drop(stream);
let rows = match conn
.query(&cx, "SELECT COUNT(*) AS count FROM streamed_cancel", &[])
.await
{
Outcome::Ok(rows) => rows,
other => panic!("connection was not returned after stream cancel: {other:?}"),
};
assert_eq!(rows[0].get_i64("count").unwrap(), 3);
});
}
#[test]
fn sqlite_file_persists_while_memory_resets_under_lab_runtime() {
init_test_logging();
let dir = tempdir().unwrap();
let db_path = dir.path().join("lab_runtime_persistence.sqlite3");
let config = TestConfig::new()
.with_seed(0x51A7_1001)
.with_tracing(true)
.with_max_steps(20_000);
let mut runtime = LabRuntimeTarget::create_runtime(config);
let (persisted_name, memory_table_count) =
LabRuntimeTarget::block_on(&mut runtime, async move {
let cx = Cx::current().expect("lab runtime should install a current Cx");
let file_conn = match SqliteConnection::open(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("file open failed: {other:?}"),
};
match file_conn
.execute_batch(
&cx,
"CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT);
INSERT INTO t(name) VALUES ('persisted');",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("file schema setup failed: {other:?}"),
}
tracing::info!(
event = %serde_json::json!({
"phase": "file_seeded",
"path": db_path.display().to_string(),
}),
"sqlite_lab_checkpoint"
);
file_conn.close().unwrap();
let reopened_file = match SqliteConnection::open(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("file reopen failed: {other:?}"),
};
let file_rows = match reopened_file.query(&cx, "SELECT name FROM t", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("file query failed after reopen: {other:?}"),
};
let persisted_name = file_rows[0].get_str("name").unwrap().to_string();
tracing::info!(
event = %serde_json::json!({
"phase": "file_reopened",
"row_count": file_rows.len(),
"name": persisted_name,
}),
"sqlite_lab_checkpoint"
);
reopened_file.close().unwrap();
let memory_conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("memory open failed: {other:?}"),
};
match memory_conn
.execute_batch(
&cx,
"CREATE TABLE ephemeral (id INTEGER PRIMARY KEY, name TEXT);
INSERT INTO ephemeral(name) VALUES ('transient');",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("memory schema setup failed: {other:?}"),
}
tracing::info!(
event = %serde_json::json!({
"phase": "memory_seeded",
"table": "ephemeral",
}),
"sqlite_lab_checkpoint"
);
memory_conn.close().unwrap();
let reopened_memory = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("memory reopen failed: {other:?}"),
};
let memory_rows = match reopened_memory
.query(
&cx,
"SELECT name FROM sqlite_master WHERE type='table' AND name='ephemeral'",
&[],
)
.await
{
Outcome::Ok(rows) => rows,
other => panic!("memory table probe failed after reopen: {other:?}"),
};
tracing::info!(
event = %serde_json::json!({
"phase": "memory_reopened",
"table_count": memory_rows.len(),
}),
"sqlite_lab_checkpoint"
);
reopened_memory.close().unwrap();
(persisted_name, memory_rows.len())
});
assert_eq!(persisted_name, "persisted");
assert_eq!(memory_table_count, 0);
let violations = runtime.oracles.check_all(runtime.now());
assert!(
violations.is_empty(),
"sqlite lab persistence test should leave runtime invariants clean: {violations:?}"
);
}
#[test]
fn sqlite_transaction_commit_persists_under_lab_runtime() {
init_test_logging();
let config = TestConfig::new()
.with_seed(0x51A7_2002)
.with_tracing(true)
.with_max_steps(20_000);
let mut runtime = LabRuntimeTarget::create_runtime(config);
let (count_inside_tx, count_after_commit, committed_name) =
LabRuntimeTarget::block_on(&mut runtime, async move {
let cx = Cx::current().expect("lab runtime should install a current Cx");
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE tx_items (id INTEGER PRIMARY KEY, name TEXT);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("schema setup failed: {other:?}"),
}
let Outcome::Ok(tx) = conn.begin(&cx).await else {
panic!("begin failed");
};
match tx
.execute(
&cx,
"INSERT INTO tx_items(name) VALUES (?1)",
&[SqliteValue::Text("committed".to_string())],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("insert in transaction failed: {other:?}"),
}
let rows_inside = match tx
.query(&cx, "SELECT COUNT(*) AS count FROM tx_items", &[])
.await
{
Outcome::Ok(rows) => rows,
other => panic!("count query inside transaction failed: {other:?}"),
};
let count_inside_tx = rows_inside[0]
.get_i64("count")
.expect("count column should be present");
tracing::info!(
event = %serde_json::json!({
"phase": "transaction_inserted",
"count_inside_tx": count_inside_tx,
}),
"sqlite_lab_checkpoint"
);
match tx.commit(&cx).await {
Outcome::Ok(()) => {}
other => panic!("commit failed: {other:?}"),
}
let rows_after = match conn
.query(
&cx,
"SELECT COUNT(*) AS count, MIN(name) AS name FROM tx_items",
&[],
)
.await
{
Outcome::Ok(rows) => rows,
other => panic!("query after commit failed: {other:?}"),
};
let count_after_commit = rows_after[0]
.get_i64("count")
.expect("count column should be present");
let committed_name = rows_after[0]
.get_str("name")
.expect("name column should be present")
.to_string();
tracing::info!(
event = %serde_json::json!({
"phase": "transaction_committed",
"count_after_commit": count_after_commit,
"name": committed_name,
}),
"sqlite_lab_checkpoint"
);
conn.close().unwrap();
(count_inside_tx, count_after_commit, committed_name)
});
assert_eq!(count_inside_tx, 1);
assert_eq!(count_after_commit, 1);
assert_eq!(committed_name, "committed");
let violations = runtime.oracles.check_all(runtime.now());
assert!(
violations.is_empty(),
"sqlite lab transaction test should leave runtime invariants clean: {violations:?}"
);
assert!(
runtime.is_quiescent(),
"lab runtime should reach quiescence"
);
}
#[test]
fn transaction_commit_cancelled_does_not_mark_finished_before_commit_runs() {
let cx = create_test_cx();
let cancelled_cx = create_test_cx();
cancelled_cx.cancel_fast(crate::types::CancelKind::User);
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
let Outcome::Ok(tx) = conn.begin(&cx).await else {
panic!("begin failed");
};
match tx.commit(&cancelled_cx).await {
Outcome::Cancelled(_) => {}
other => panic!("expected cancelled commit, got: {other:?}"),
}
for _ in 0..8 {
if conn
.inner
.lock()
.get()
.is_ok_and(rusqlite::Connection::is_autocommit)
{
break;
}
match conn.query(&cx, "SELECT 1", &[]).await {
Outcome::Ok(_) => {}
other => panic!("probe query failed: {other:?}"),
}
}
assert!(
conn.inner
.lock()
.get()
.is_ok_and(rusqlite::Connection::is_autocommit),
"connection should return to autocommit after cancelled commit drop path"
);
});
}
#[test]
fn open_file_sets_wal_mode() {
let cx = create_test_cx();
let dir = tempdir().unwrap();
let db_path = dir.path().join("wal_mode.sqlite3");
block_on(async {
let conn = match SqliteConnection::open(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("open failed: {other:?}"),
};
let rows = match conn.query_unchecked(&cx, "PRAGMA journal_mode", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("query pragma failed: {other:?}"),
};
let mode = rows[0]
.get_idx(0)
.unwrap()
.as_text()
.unwrap()
.to_ascii_lowercase();
assert_eq!(mode, "wal");
});
}
#[test]
fn sqlite_value_invalid_utf8_is_typed_and_recovers_across_query_surfaces() {
let cx = create_test_cx();
block_on(async {
let mut conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
const INVALID_TEXT_QUERY: &str = "SELECT CAST(X'80' AS TEXT) AS bad_text";
match conn.query_unchecked(&cx, INVALID_TEXT_QUERY, &[]).await {
Outcome::Err(SqliteError::InvalidTextEncoding { column, .. }) => {
assert_eq!(column, "bad_text")
}
other => panic!("expected invalid UTF-8 rejection, got: {other:?}"),
}
match conn.query_row_unchecked(&cx, INVALID_TEXT_QUERY, &[]).await {
Outcome::Err(SqliteError::InvalidTextEncoding { column, .. }) => {
assert_eq!(column, "bad_text")
}
other => panic!("expected query_row invalid UTF-8 rejection, got: {other:?}"),
}
let mut stream = match conn
.query_stream_unchecked(&cx, INVALID_TEXT_QUERY, &[])
.await
{
Outcome::Ok(stream) => stream,
other => panic!("query_stream failed to start: {other:?}"),
};
match stream.next(&cx).await {
Outcome::Err(SqliteError::InvalidTextEncoding { column, .. }) => {
assert_eq!(column, "bad_text")
}
other => panic!("expected streamed invalid UTF-8 rejection, got: {other:?}"),
}
drop(stream);
match conn.query_unchecked(&cx, "SELECT 1 AS healthy", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows[0].get_i64("healthy").unwrap(), 1),
other => panic!("connection did not recover after UTF-8 errors: {other:?}"),
}
});
}
#[test]
fn unchecked_execute_rejects_attach_database() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_unchecked(&cx, "ATTACH ':memory:' AS audit", &[])
.await
{
Outcome::Err(SqliteError::UnsafeSql(msg)) => {
assert!(msg.contains("ATTACH and DETACH"));
}
other => panic!("expected ATTACH rejection, got: {other:?}"),
}
});
}
#[test]
fn open_rejects_tilde_prefixed_paths_before_rusqlite() {
let cx = create_test_cx();
block_on(async {
match SqliteConnection::open(&cx, "~/tenant.sqlite").await {
Outcome::Err(SqliteError::UnsafePath(msg)) => {
assert!(msg.contains("tilde-prefixed"));
}
other => panic!("expected unsafe path rejection, got: {other:?}"),
}
});
}
#[test]
fn open_rejects_parent_directory_traversal_before_rusqlite() {
let cx = create_test_cx();
block_on(async {
match SqliteConnection::open(&cx, "../tenant.sqlite").await {
Outcome::Err(SqliteError::UnsafePath(msg)) => {
assert!(msg.contains("parent-directory traversal"));
}
other => panic!("expected unsafe traversal rejection, got: {other:?}"),
}
});
}
#[test]
fn separate_validated_connections_keep_schema_isolated_without_attach() {
let cx = create_test_cx();
block_on(async {
let dir = tempfile::tempdir().expect("tempdir");
let first_path = dir.path().join("tenant_a.sqlite3");
let second_path = dir.path().join("tenant_b.sqlite3");
let first = match SqliteConnection::open(&cx, &first_path).await {
Outcome::Ok(conn) => conn,
other => panic!("open first db failed: {other:?}"),
};
let second = match SqliteConnection::open(&cx, &second_path).await {
Outcome::Ok(conn) => conn,
other => panic!("open second db failed: {other:?}"),
};
match first
.execute_batch(
&cx,
"CREATE TABLE tenant_only (id INTEGER PRIMARY KEY, value TEXT);
INSERT INTO tenant_only(value) VALUES ('a');",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("seed first db failed: {other:?}"),
}
let rows = match second
.query(
&cx,
"SELECT name FROM sqlite_master WHERE type='table' AND name='tenant_only'",
&[],
)
.await
{
Outcome::Ok(rows) => rows,
other => panic!("query second db failed: {other:?}"),
};
assert!(
rows.is_empty(),
"separate validated sqlite connections must not share attached schema state"
);
});
}
#[test]
fn sqlite_rowid_max_round_trips_without_overflow() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
match conn
.execute(
&cx,
"INSERT INTO t(id, name) VALUES (?1, ?2)",
&[
SqliteValue::Integer(i64::MAX),
SqliteValue::Text("max-rowid".to_string()),
],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("insert failed: {other:?}"),
}
let rows = match conn
.query(&cx, "SELECT rowid AS rowid, id, name FROM t", &[])
.await
{
Outcome::Ok(rows) => rows,
other => panic!("query failed: {other:?}"),
};
assert_eq!(rows[0].get_i64("rowid").unwrap(), i64::MAX);
assert_eq!(rows[0].get_i64("id").unwrap(), i64::MAX);
assert_eq!(rows[0].get_str("name").unwrap(), "max-rowid");
});
}
#[test]
fn sqlite_rowid_overflow_literal_is_rejected() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
match conn
.execute_unchecked(
&cx,
"INSERT INTO t(id, name) VALUES(9223372036854775808, 'overflow')",
&[],
)
.await
{
Outcome::Err(SqliteError::Sqlite(msg)) => {
assert!(
msg.to_ascii_lowercase().contains("datatype mismatch"),
"unexpected rowid overflow error: {msg}"
);
}
other => panic!("expected rowid overflow rejection, got: {other:?}"),
}
});
}
#[test]
fn transaction_drop_rolls_back_uncommitted_work() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
let Outcome::Ok(tx) = conn.begin(&cx).await else {
panic!("begin failed");
};
match tx
.execute(
&cx,
"INSERT INTO t(v) VALUES (?1)",
&[SqliteValue::Text("x".to_string())],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("insert in tx failed: {other:?}"),
}
drop(tx);
let rows = match conn.query(&cx, "SELECT COUNT(*) FROM t", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("count query failed: {other:?}"),
};
assert_eq!(rows[0].get_idx(0).unwrap().as_integer(), Some(0));
});
}
#[test]
fn dropped_transaction_rolls_back_before_followup_connection_operation() {
let cx = create_test_cx();
let pool = BlockingPool::new(1, 1);
let raw = rusqlite::Connection::open_in_memory().expect("open test connection");
configure_connection_defaults(&raw, false).expect("configure test connection");
let interrupt = Arc::new(raw.get_interrupt_handle());
let conn = SqliteConnection {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(raw))),
pool: pool.handle(),
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
};
block_on(async {
match conn
.execute_batch(&cx, "CREATE TABLE t (value INTEGER NOT NULL);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
let transaction = match conn.begin_immediate(&cx).await {
Outcome::Ok(transaction) => transaction,
other => panic!("begin immediate failed with {:?}", other.severity()),
};
match transaction
.execute(&cx, "INSERT INTO t (value) VALUES (1)", &[])
.await
{
Outcome::Ok(1) => {}
other => panic!("insert in transaction failed: {other:?}"),
}
drop(transaction);
});
conn.pool.spawn(|| {}).wait();
let guard = conn.inner.lock();
let raw = guard.get().expect("connection remains open");
assert!(
raw.is_autocommit(),
"drop-triggered cleanup must end the physical transaction"
);
let retained_rows: i64 = raw
.query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0))
.expect("read direct post-drop row count");
assert_eq!(retained_rows, 0, "uncommitted row must be rolled back");
drop(guard);
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::Autocommit,
"drop-triggered cleanup must restore the transaction mirror"
);
}
#[test]
fn delayed_drop_cleanup_does_not_rollback_newer_transaction_generation() {
let cx = create_test_cx();
let pool = BlockingPool::new(1, 1);
let raw = rusqlite::Connection::open_in_memory().expect("open test connection");
configure_connection_defaults(&raw, false).expect("configure test connection");
raw.execute_batch("CREATE TABLE t (value INTEGER NOT NULL)")
.expect("create test table");
let interrupt = Arc::new(raw.get_interrupt_handle());
let conn = SqliteConnection {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(raw))),
pool: pool.handle(),
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
};
let transaction = match block_on(conn.begin_immediate(&cx)) {
Outcome::Ok(transaction) => transaction,
other => panic!("begin immediate failed with {:?}", other.severity()),
};
match block_on(transaction.execute(&cx, "INSERT INTO t (value) VALUES (1)", &[])) {
Outcome::Ok(1) => {}
other => panic!("insert in transaction failed: {other:?}"),
}
let inner = Arc::clone(&conn.inner);
let guard = inner.lock();
let old_generation = transaction.generation;
drop(transaction);
let raw = guard.get().expect("connection remains open");
raw.execute_batch("ROLLBACK; BEGIN IMMEDIATE; INSERT INTO t (value) VALUES (2)")
.expect("install replacement transaction");
let replacement_generation =
advance_transaction_generation(conn.transaction_generation.as_ref())
.expect("advance replacement generation");
assert_ne!(replacement_generation, old_generation);
*conn.transaction_state.lock() = TransactionState::InTransaction;
drop(guard);
conn.pool.spawn(|| {}).wait();
let guard = conn.inner.lock();
let raw = guard.get().expect("connection remains open");
assert!(
!raw.is_autocommit(),
"stale drop cleanup must not finish the replacement transaction"
);
let visible_rows: i64 = raw
.query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0))
.expect("read replacement transaction row count");
assert_eq!(visible_rows, 1);
raw.execute_batch("ROLLBACK")
.expect("clean up replacement transaction");
drop(guard);
*conn.transaction_state.lock() = TransactionState::Autocommit;
}
#[test]
fn hard_dropped_begin_recovers_after_cleanup_overtakes_worker() {
let cx = create_test_cx();
let pool = BlockingPool::new(1, 1);
let raw = rusqlite::Connection::open_in_memory().expect("open test connection");
configure_connection_defaults(&raw, false).expect("configure test connection");
let interrupt = Arc::new(raw.get_interrupt_handle());
let conn = SqliteConnection {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(raw))),
pool: pool.handle(),
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
};
macro_rules! assert_hard_drop_recovers {
($begin:expr, $mode:literal) => {{
let inner = Arc::clone(&conn.inner);
let inner_guard = inner.lock();
let waker = std::task::Waker::noop();
let mut task_cx = std::task::Context::from_waker(waker);
let mut begin = Box::pin($begin);
assert!(
std::future::Future::poll(begin.as_mut(), &mut task_cx).is_pending(),
"{} waits behind the held connection mutex",
$mode
);
drop(begin);
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::NeedsRollback,
"hard-dropping {} must poison the transaction mirror",
$mode
);
rollback_orphaned_transaction_mutex_guarded(
inner_guard.get().expect("connection remains open"),
conn.transaction_state.as_ref(),
)
.expect("overtaking cleanup succeeds");
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::Autocommit,
"overtaking cleanup clears the initial poison"
);
drop(inner_guard);
conn.pool.spawn(|| {}).wait();
let inner_guard = conn.inner.lock();
assert!(
inner_guard
.get()
.expect("connection remains open")
.is_autocommit(),
"abandoned {} must not leave a physical transaction open",
$mode
);
drop(inner_guard);
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::Autocommit,
"abandoned {} must restore the transaction mirror",
$mode
);
}};
}
assert_hard_drop_recovers!(conn.begin(&cx), "BEGIN");
assert_hard_drop_recovers!(conn.begin_immediate(&cx), "BEGIN IMMEDIATE");
assert_hard_drop_recovers!(conn.begin_exclusive(&cx), "BEGIN EXCLUSIVE");
}
#[test]
fn hard_dropped_begin_after_worker_completion_survives_lagging_commit() {
let cx = create_test_cx();
let pool = BlockingPool::new(1, 1);
let raw = rusqlite::Connection::open_in_memory().expect("open test connection");
configure_connection_defaults(&raw, false).expect("configure test connection");
let interrupt = Arc::new(raw.get_interrupt_handle());
let conn = SqliteConnection {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(raw))),
pool: pool.handle(),
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
};
let transaction = match block_on(conn.begin(&cx)) {
Outcome::Ok(transaction) => transaction,
other => panic!("initial BEGIN failed with {:?}", other.severity()),
};
let inner = Arc::clone(&conn.inner);
let inner_guard = inner.lock();
let waker = std::task::Waker::noop();
let mut task_cx = std::task::Context::from_waker(waker);
let mut commit = Box::pin(transaction.commit(&cx));
let mut begin = Box::pin(conn.begin(&cx));
assert!(
std::future::Future::poll(commit.as_mut(), &mut task_cx).is_pending(),
"COMMIT waits behind the held connection mutex"
);
assert!(
std::future::Future::poll(begin.as_mut(), &mut task_cx).is_pending(),
"BEGIN queues behind COMMIT"
);
drop(inner_guard);
conn.pool.spawn(|| {}).wait();
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::InTransaction,
"the second BEGIN is physically open before its consumer resumes"
);
drop(commit);
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::InTransaction,
"completed COMMIT future drop must not poison the newer BEGIN"
);
drop(begin);
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::NeedsRollback,
"late BEGIN drop must poison an already-opened transaction"
);
match block_on(conn.set_busy_timeout(&cx, Duration::ZERO)) {
Outcome::Ok(()) => {}
other => panic!("post-drop cleanup failed: {other:?}"),
}
let inner_guard = conn.inner.lock();
assert!(
inner_guard
.get()
.expect("connection remains open")
.is_autocommit(),
"the next operation must drain the late-dropped BEGIN"
);
drop(inner_guard);
assert_eq!(*conn.transaction_state.lock(), TransactionState::Autocommit);
}
#[test]
fn stale_finish_worker_skips_newer_transaction_after_cleanup_overtakes() {
struct BarrierReleaseGuard(Option<Arc<std::sync::Barrier>>);
impl BarrierReleaseGuard {
fn release(&mut self) {
if let Some(barrier) = self.0.take() {
barrier.wait();
}
}
}
impl Drop for BarrierReleaseGuard {
fn drop(&mut self) {
self.release();
}
}
let cx = create_test_cx();
for (finish_sql, kind) in [
("COMMIT", TransactionFinishKind::Commit),
("ROLLBACK", TransactionFinishKind::Rollback),
] {
let pool = BlockingPool::new(2, 2);
let raw = rusqlite::Connection::open_in_memory().expect("open test connection");
configure_connection_defaults(&raw, false).expect("configure test connection");
raw.execute_batch("CREATE TABLE replacement_rows (value INTEGER NOT NULL)")
.expect("create test table");
let interrupt = Arc::new(raw.get_interrupt_handle());
let conn = SqliteConnection {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(raw))),
pool: pool.handle(),
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
};
let mut old_transaction = match block_on(conn.begin(&cx)) {
Outcome::Ok(transaction) => transaction,
other => panic!("initial BEGIN failed with {:?}", other.severity()),
};
let mut stale_finish = TransactionFinishEffect::new(
Arc::clone(&conn.transaction_state),
Arc::clone(&conn.transaction_generation),
old_transaction.generation,
kind,
old_transaction.obligation.take(),
);
let entered = Arc::new(std::sync::Barrier::new(2));
let release = Arc::new(std::sync::Barrier::new(2));
let worker_entered = Arc::clone(&entered);
let worker_release = Arc::clone(&release);
let stale_inner = Arc::clone(&conn.inner);
let (result_tx, result_rx) = std::sync::mpsc::sync_channel(1);
let stale_handle = conn.pool.spawn(move || {
worker_entered.wait();
worker_release.wait();
let guard = stale_inner.lock();
let result = stale_finish
.execute_worker(guard.get().expect("connection remains open"), finish_sql);
result_tx.send(result).expect("publish stale worker result");
});
entered.wait();
let mut release_on_unwind = BarrierReleaseGuard(Some(Arc::clone(&release)));
drop(old_transaction);
let replacement = match block_on(conn.begin(&cx)) {
Outcome::Ok(transaction) => transaction,
other => panic!("replacement BEGIN failed with {:?}", other.severity()),
};
match block_on(replacement.execute(
&cx,
"INSERT INTO replacement_rows (value) VALUES (1)",
&[],
)) {
Outcome::Ok(1) => {}
other => panic!("replacement INSERT failed: {other:?}"),
}
let replacement_generation = replacement.generation;
release_on_unwind.release();
stale_handle.wait();
let stale_result = result_rx.recv().expect("receive stale worker result");
assert!(matches!(
stale_result,
Err(SqliteError::TransactionFinished)
));
assert_eq!(
conn.transaction_generation.load(Ordering::Acquire),
replacement_generation,
"stale {finish_sql} must not advance the replacement generation"
);
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::InTransaction
);
{
let guard = conn.inner.lock();
assert!(
!guard
.get()
.expect("connection remains open")
.is_autocommit(),
"stale {finish_sql} must leave the replacement transaction open"
);
}
let finish_outcome = match kind {
TransactionFinishKind::Commit => block_on(replacement.rollback(&cx)),
TransactionFinishKind::Rollback => block_on(replacement.commit(&cx)),
};
assert!(matches!(finish_outcome, Outcome::Ok(())));
let rows = match block_on(conn.query(
&cx,
"SELECT COUNT(*) AS count FROM replacement_rows",
&[],
)) {
Outcome::Ok(rows) => rows,
other => panic!("replacement row count failed: {other:?}"),
};
let expected_rows = match kind {
TransactionFinishKind::Commit => 0,
TransactionFinishKind::Rollback => 1,
};
assert_eq!(
rows[0].get_i64("count").expect("read replacement count"),
expected_rows,
"stale {finish_sql} must not finish the replacement transaction"
);
}
}
#[test]
fn managed_begin_fails_closed_before_generation_exhaustion() {
let raw = rusqlite::Connection::open_in_memory().expect("open test connection");
configure_connection_defaults(&raw, false).expect("configure test connection");
let transaction_state = Arc::new(Mutex::new(TransactionState::Autocommit));
let transaction_generation = Arc::new(AtomicU64::new(u64::MAX - 1));
let begin = TransactionWorkerEffect::Begin(BeginAttempt::new(
Arc::clone(&transaction_state),
Arc::clone(&transaction_generation),
));
let result = begin.execute_worker(&raw, "BEGIN");
assert!(matches!(result, Err(SqliteError::Sqlite(_))));
assert!(raw.is_autocommit(), "exhausted BEGIN must issue no SQL");
assert_eq!(transaction_generation.load(Ordering::Acquire), u64::MAX - 1);
assert_eq!(*transaction_state.lock(), TransactionState::Autocommit);
}
#[test]
fn transaction_drop_preserves_foreign_key_cascade_consistency() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"
CREATE TABLE parent (id INTEGER PRIMARY KEY);
CREATE TABLE child (
id INTEGER PRIMARY KEY,
parent_id INTEGER NOT NULL REFERENCES parent(id) ON DELETE CASCADE
);
INSERT INTO parent(id) VALUES (1);
INSERT INTO child(id, parent_id) VALUES (10, 1);
",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("schema setup failed: {other:?}"),
}
let Outcome::Ok(tx) = conn.begin_immediate(&cx).await else {
panic!("begin_immediate failed");
};
match tx
.execute(&cx, "DELETE FROM parent WHERE id = 1", &[])
.await
{
Outcome::Ok(1) => {}
other => panic!("delete in transaction failed: {other:?}"),
}
drop(tx);
let parent_rows = match conn.query(&cx, "SELECT COUNT(*) FROM parent", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("parent count failed: {other:?}"),
};
let child_rows = match conn.query(&cx, "SELECT COUNT(*) FROM child", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("child count failed: {other:?}"),
};
assert_eq!(parent_rows[0].get_idx(0).unwrap().as_integer(), Some(1));
assert_eq!(child_rows[0].get_idx(0).unwrap().as_integer(), Some(1));
match conn
.execute(&cx, "DELETE FROM parent WHERE id = 1", &[])
.await
{
Outcome::Ok(1) => {}
other => panic!("post-rollback delete failed: {other:?}"),
}
let child_rows = match conn.query(&cx, "SELECT COUNT(*) FROM child", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("child recount failed: {other:?}"),
};
assert_eq!(child_rows[0].get_idx(0).unwrap().as_integer(), Some(0));
});
}
#[test]
fn sqlite_prepared_statement_cache_capacity_one_reuses_evicts_and_reprepares() {
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
{
let guard = conn.inner.lock();
let raw = guard.get().expect("connection open");
raw.set_prepared_statement_cache_capacity(1);
}
match conn
.execute_batch(
&cx,
"
CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT);
INSERT INTO t(value) VALUES ('before');
",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("initial schema setup failed: {other:?}"),
}
match conn
.query(&cx, "SELECT value FROM t WHERE id = 1", &[])
.await
{
Outcome::Ok(rows) => assert_eq!(rows[0].get_str("value").unwrap(), "before"),
other => panic!("initial cached query failed: {other:?}"),
}
match conn
.query(&cx, "SELECT value FROM t WHERE id = 1", &[])
.await
{
Outcome::Ok(rows) => assert_eq!(rows[0].get_str("value").unwrap(), "before"),
other => panic!("cached query reuse failed: {other:?}"),
}
match conn.query(&cx, "SELECT id FROM t WHERE id = 1", &[]).await {
Outcome::Ok(rows) => assert_eq!(rows[0].get_i64("id").unwrap(), 1),
other => panic!("second cached query failed: {other:?}"),
}
match conn
.execute_batch(
&cx,
"
DROP TABLE t;
CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT);
INSERT INTO t(value) VALUES ('after');
",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("schema rebuild failed: {other:?}"),
}
match conn
.query(&cx, "SELECT value FROM t WHERE id = 1", &[])
.await
{
Outcome::Ok(rows) => assert_eq!(rows[0].get_str("value").unwrap(), "after"),
other => panic!("cached query after schema change failed: {other:?}"),
}
});
}
#[test]
fn busy_timeout_produces_lock_error_under_write_contention() {
let cx = create_test_cx();
let dir = tempdir().unwrap();
let db_path = dir.path().join("busy_timeout.sqlite3");
block_on(async {
let conn1 = match SqliteConnection::open(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("open conn1 failed: {other:?}"),
};
let conn2 = match SqliteConnection::open(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("open conn2 failed: {other:?}"),
};
match conn1
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
match conn2.set_busy_timeout(&cx, Duration::from_millis(50)).await {
Outcome::Ok(()) => {}
other => panic!("set_busy_timeout failed: {other:?}"),
}
let Outcome::Ok(tx) = conn1.begin_immediate(&cx).await else {
panic!("begin_immediate failed");
};
match conn2
.execute(
&cx,
"INSERT INTO t(v) VALUES (?1)",
&[SqliteValue::Text("blocked".to_string())],
)
.await
{
Outcome::Err(SqliteError::Sqlite(msg)) => {
let lower = msg.to_ascii_lowercase();
assert!(
lower.contains("database is locked") || lower.contains("database is busy"),
"unexpected busy error message: {msg}"
);
}
other => panic!("expected lock error, got: {other:?}"),
}
match tx.rollback(&cx).await {
Outcome::Ok(()) => {}
other => panic!("rollback failed: {other:?}"),
}
});
}
#[test]
fn sqlite_p8_engine_codes_map_without_rendered_message_parsing() {
let cases = [
(
rusqlite::ffi::SQLITE_BUSY,
SqliteErrorCategory::Busy,
"SQLITE_BUSY",
SqliteRetryDisposition::RetryOperation,
false,
),
(
rusqlite::ffi::SQLITE_LOCKED,
SqliteErrorCategory::Locked,
"SQLITE_LOCKED",
SqliteRetryDisposition::RetryOperation,
false,
),
(
rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE,
SqliteErrorCategory::Constraint,
"SQLITE_CONSTRAINT",
SqliteRetryDisposition::Never,
false,
),
(
rusqlite::ffi::SQLITE_INTERRUPT,
SqliteErrorCategory::Interrupted,
"SQLITE_INTERRUPT",
SqliteRetryDisposition::Never,
false,
),
(
rusqlite::ffi::SQLITE_IOERR_READ,
SqliteErrorCategory::Io,
"SQLITE_IOERR",
SqliteRetryDisposition::ReopenConnection,
true,
),
(
rusqlite::ffi::SQLITE_NOTADB,
SqliteErrorCategory::Corrupt,
"SQLITE_NOTADB",
SqliteRetryDisposition::ReopenConnection,
true,
),
(
rusqlite::ffi::SQLITE_AUTH,
SqliteErrorCategory::PermissionDenied,
"SQLITE_AUTH",
SqliteRetryDisposition::Never,
false,
),
];
for (extended, category, primary, retry, connection_error) in cases {
let error = SqliteOperationError::from_rusqlite(
SqliteOperation::Step,
rusqlite::Error::SqliteFailure(
rusqlite::ffi::Error::new(extended),
Some("sensitive SQL and path payload".to_owned()),
),
);
let diagnostic = error.diagnostic();
assert_eq!(diagnostic.operation(), SqliteOperation::Step);
assert_eq!(diagnostic.category(), category);
assert_eq!(diagnostic.primary_code(), Some(primary));
assert_eq!(diagnostic.extended_code(), Some(extended));
assert_eq!(diagnostic.retry_disposition(), retry);
assert_eq!(diagnostic.is_connection_error(), connection_error);
assert_eq!(
diagnostic.is_retryable(),
retry == SqliteRetryDisposition::RetryOperation
);
let rendered = format!("{error:?} {error}");
assert!(!rendered.contains("sensitive SQL and path payload"));
assert!(error.engine_source().is_some());
assert!(std::error::Error::source(&error).is_none());
assert!(
matches!(error.legacy_error(), SqliteError::Sqlite(message) if message.contains("sensitive SQL and path payload")),
"the legacy source remains available only through the explicit accessor"
);
}
let path_error = SqliteOperationError::from_legacy(
SqliteOperation::Validation,
SqliteError::UnsafePath("sensitive path".to_owned()),
);
assert_eq!(
path_error.diagnostic().category(),
SqliteErrorCategory::PermissionDenied
);
assert_eq!(
path_error.diagnostic().operator_code(),
"sqlite.permission_denied"
);
let missing_column = SqliteOperationError::from_legacy(
SqliteOperation::Step,
SqliteError::ColumnNotFound("secret_column".to_owned()),
);
assert_eq!(
missing_column.diagnostic().category(),
SqliteErrorCategory::NotFound
);
assert!(!format!("{missing_column:?}").contains("secret_column"));
}
#[test]
fn sqlite_p8_public_diagnosed_apis_preserve_legacy_and_reuse() {
let cx = create_test_cx();
let directory = tempdir().unwrap();
let directory_path = directory.path().to_path_buf();
block_on(async {
let open_error = match SqliteConnection::open_diagnosed(&cx, &directory_path).await {
Outcome::Err(error) => error,
Outcome::Ok(connection) => {
drop(connection);
panic!("diagnosed directory open unexpectedly succeeded")
}
Outcome::Cancelled(reason) => {
panic!("diagnosed directory open was cancelled: {reason:?}")
}
Outcome::Panicked(_) => panic!("diagnosed directory open panicked"),
};
assert_eq!(open_error.diagnostic().operation(), SqliteOperation::Open);
assert_eq!(open_error.diagnostic().category(), SqliteErrorCategory::Io);
assert_eq!(
open_error.diagnostic().primary_code(),
Some("SQLITE_CANTOPEN")
);
let conn = match SqliteConnection::open_in_memory_diagnosed(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("diagnosed open failed: {other:?}"),
};
match conn
.execute_batch_diagnosed(
&cx,
"CREATE TABLE p8 (id INTEGER PRIMARY KEY, value TEXT UNIQUE NOT NULL);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("diagnosed schema setup failed: {other:?}"),
}
match conn
.execute_diagnosed(
&cx,
"INSERT INTO p8(id, value) VALUES (?1, ?2)",
&[
SqliteValue::Integer(1),
SqliteValue::Text("first".to_owned()),
],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("diagnosed insert failed: {other:?}"),
}
let prepare = match conn
.query_unchecked_diagnosed(&cx, "SELEKT p8_sensitive_payload", &[])
.await
{
Outcome::Err(error) => error,
other => panic!("expected diagnosed prepare failure, got {other:?}"),
};
assert_eq!(prepare.diagnostic().operation(), SqliteOperation::Prepare);
assert_eq!(
prepare.diagnostic().category(),
SqliteErrorCategory::InvalidInput
);
assert_eq!(prepare.diagnostic().primary_code(), Some("SQLITE_ERROR"));
assert!(!format!("{prepare:?} {prepare}").contains("p8_sensitive_payload"));
let duplicate = match conn
.execute_diagnosed(
&cx,
"INSERT INTO p8(id, value) VALUES (?1, ?2)",
&[
SqliteValue::Integer(2),
SqliteValue::Text("first".to_owned()),
],
)
.await
{
Outcome::Err(error) => error,
other => panic!("expected diagnosed constraint failure, got {other:?}"),
};
assert_eq!(duplicate.diagnostic().operation(), SqliteOperation::Step);
assert_eq!(
duplicate.diagnostic().category(),
SqliteErrorCategory::Constraint
);
assert_eq!(
duplicate.diagnostic().primary_code(),
Some("SQLITE_CONSTRAINT")
);
assert!(!duplicate.diagnostic().is_retryable());
assert!(matches!(duplicate.legacy_error(), SqliteError::Sqlite(_)));
let bind = match conn
.execute_diagnosed(
&cx,
"INSERT INTO p8(id, value) VALUES (?1, ?2)",
&[SqliteValue::Integer(3)],
)
.await
{
Outcome::Err(error) => error,
other => panic!("expected diagnosed bind failure, got {other:?}"),
};
assert_eq!(bind.diagnostic().operation(), SqliteOperation::Bind);
assert_eq!(
bind.diagnostic().category(),
SqliteErrorCategory::InvalidInput
);
assert_eq!(bind.diagnostic().primary_code(), None);
let rejected = match conn.query_diagnosed(&cx, "PRAGMA journal_mode", &[]).await {
Outcome::Err(error) => error,
other => panic!("checked policy must reject PRAGMA, got {other:?}"),
};
assert_eq!(
rejected.diagnostic().operation(),
SqliteOperation::Validation
);
assert_eq!(
rejected.diagnostic().category(),
SqliteErrorCategory::InvalidInput
);
let transaction = match conn.begin_diagnosed(&cx).await {
Outcome::Ok(transaction) => transaction,
Outcome::Err(error) => panic!("diagnosed begin failed: {error}"),
Outcome::Cancelled(reason) => {
panic!("diagnosed begin was cancelled: {reason:?}")
}
Outcome::Panicked(_) => panic!("diagnosed begin panicked"),
};
let nested = match conn.begin_diagnosed(&cx).await {
Outcome::Err(error) => error,
Outcome::Ok(_) => panic!("nested diagnosed transaction unexpectedly began"),
Outcome::Cancelled(reason) => {
panic!("nested diagnosed begin was cancelled: {reason:?}")
}
Outcome::Panicked(_) => panic!("nested diagnosed begin panicked"),
};
assert_eq!(
nested.diagnostic().operation(),
SqliteOperation::TransactionBegin
);
assert_eq!(
nested.diagnostic().category(),
SqliteErrorCategory::InvalidInput
);
match transaction
.execute_diagnosed(
&cx,
"INSERT INTO p8(id, value) VALUES (?1, ?2)",
&[
SqliteValue::Integer(4),
SqliteValue::Text("rolled_back".to_owned()),
],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("diagnosed transaction insert failed: {other:?}"),
}
match transaction.rollback_diagnosed(&cx).await {
Outcome::Ok(()) => {}
other => panic!("diagnosed rollback failed: {other:?}"),
}
match conn
.query_row_diagnosed(
&cx,
"SELECT COUNT(*) AS count FROM p8 WHERE value = ?1",
&[SqliteValue::Text("rolled_back".to_owned())],
)
.await
{
Outcome::Ok(Some(row)) => assert_eq!(row.get_i64("count").unwrap(), 0),
other => panic!("connection was not reusable after errors: {other:?}"),
}
match conn.close_async_diagnosed(&cx).await {
Outcome::Ok(()) => {}
other => panic!("diagnosed close failed: {other:?}"),
}
assert!(!conn.is_open());
let closed = match conn.query_unchecked_diagnosed(&cx, "SELECT 1", &[]).await {
Outcome::Err(error) => error,
other => panic!("closed diagnosed connection accepted query: {other:?}"),
};
assert_eq!(closed.diagnostic().category(), SqliteErrorCategory::Closed);
assert_eq!(
closed.diagnostic().retry_disposition(),
SqliteRetryDisposition::ReopenConnection
);
});
}
#[test]
fn sqlite_p8_busy_cancel_interrupt_and_pool_shutdown_are_distinct() {
let cx = create_test_cx();
let dir = tempdir().unwrap();
let db_path = dir.path().join("p8_contention.sqlite3");
block_on(async {
let conn1 = match SqliteConnection::open_diagnosed(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("diagnosed open conn1 failed: {other:?}"),
};
let conn2 = match SqliteConnection::open_diagnosed(&cx, &db_path).await {
Outcome::Ok(conn) => conn,
other => panic!("diagnosed open conn2 failed: {other:?}"),
};
match conn1
.execute_batch_diagnosed(
&cx,
"CREATE TABLE p8_busy (id INTEGER PRIMARY KEY, value TEXT);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("busy schema setup failed: {other:?}"),
}
match conn2
.set_busy_timeout_diagnosed(&cx, Duration::from_millis(25))
.await
{
Outcome::Ok(()) => {}
other => panic!("diagnosed busy timeout failed: {other:?}"),
}
let transaction = match conn1.begin_immediate_diagnosed(&cx).await {
Outcome::Ok(transaction) => transaction,
Outcome::Err(error) => panic!("diagnosed immediate begin failed: {error}"),
Outcome::Cancelled(reason) => {
panic!("diagnosed immediate begin was cancelled: {reason:?}")
}
Outcome::Panicked(_) => panic!("diagnosed immediate begin panicked"),
};
let busy = match conn2
.execute_diagnosed(
&cx,
"INSERT INTO p8_busy(value) VALUES (?1)",
&[SqliteValue::Text("blocked".to_owned())],
)
.await
{
Outcome::Err(error) => error,
other => panic!("expected diagnosed busy failure, got {other:?}"),
};
assert!(matches!(
busy.diagnostic().category(),
SqliteErrorCategory::Busy | SqliteErrorCategory::Locked
));
assert!(busy.diagnostic().is_retryable());
assert!(!busy.diagnostic().is_connection_error());
match transaction.rollback_diagnosed(&cx).await {
Outcome::Ok(()) => {}
other => panic!("diagnosed contention rollback failed: {other:?}"),
}
match conn2
.execute_diagnosed(
&cx,
"INSERT INTO p8_busy(value) VALUES (?1)",
&[SqliteValue::Text("recovered".to_owned())],
)
.await
{
Outcome::Ok(1) => {}
other => panic!("contender was not reusable: {other:?}"),
}
});
let pool = BlockingPool::new(1, 1);
let raw = rusqlite::Connection::open_in_memory().expect("open dedicated P8 connection");
configure_connection_defaults(&raw, false).expect("configure dedicated P8 connection");
let interrupt = Arc::new(raw.get_interrupt_handle());
let conn = SqliteConnection {
inner: Arc::new(Mutex::new(SqliteConnectionInner::new(raw))),
pool: pool.handle(),
transaction_state: Arc::new(Mutex::new(TransactionState::Autocommit)),
transaction_generation: Arc::new(AtomicU64::new(0)),
interrupt,
statement_timeout_override: None,
};
let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1);
let mut interrupted = Box::pin(conn.run_connection_op_diagnosed(
&cx,
"sqlite P8 explicit interrupt",
SqliteOperation::Step,
move |raw| run_signalled_infinite_query_diagnosed(raw, started_tx),
));
assert!(
block_on(futures_lite::future::poll_once(interrupted.as_mut())).is_none(),
"infinite operation must park before explicit interrupt"
);
started_rx
.recv_timeout(Duration::from_secs(5))
.expect("diagnosed statement must start");
conn.interrupt();
match block_on(interrupted) {
Outcome::Err(error) => {
assert_eq!(
error.diagnostic().category(),
SqliteErrorCategory::Interrupted
);
assert_eq!(error.diagnostic().primary_code(), Some("SQLITE_INTERRUPT"));
}
other => panic!("explicit interrupt must remain an error: {other:?}"),
}
let cancelled_cx = create_test_cx();
let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1);
let mut cancelled = Box::pin(conn.run_connection_op_diagnosed(
&cancelled_cx,
"sqlite P8 cancellation",
SqliteOperation::Step,
move |raw| run_signalled_infinite_query_diagnosed(raw, started_tx),
));
assert!(
block_on(futures_lite::future::poll_once(cancelled.as_mut())).is_none(),
"infinite operation must park before Cx cancellation"
);
started_rx
.recv_timeout(Duration::from_secs(5))
.expect("cancellable diagnosed statement must start");
cancelled_cx.cancel_fast(crate::types::CancelKind::User);
match block_on(cancelled) {
Outcome::Cancelled(reason) => {
assert_eq!(reason.kind, crate::types::CancelKind::User);
}
other => panic!("Cx cancellation must not become an error category: {other:?}"),
}
let fresh_cx = create_test_cx();
match block_on(conn.query_unchecked_diagnosed(&fresh_cx, "SELECT 1", &[])) {
Outcome::Ok(rows) => assert_eq!(rows.len(), 1),
other => panic!("connection unusable after interrupt/cancel: {other:?}"),
}
conn.close_diagnosed()
.expect("close dedicated P8 connection");
drop(conn);
assert!(
pool.shutdown_and_wait(Duration::from_secs(5)),
"dedicated P8 blocking pool must shut down"
);
assert_eq!(pool.pending_count(), 0);
assert_eq!(pool.busy_threads(), 0);
assert_eq!(pool.active_threads(), 0);
}
#[test]
fn execute_with_cancelled_cx_does_not_mutate_state() {
let cx = create_test_cx();
let cancelled = create_test_cx();
cancelled.cancel_fast(crate::types::CancelKind::User);
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(&cx, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT);")
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
match conn
.execute(
&cancelled,
"INSERT INTO t(v) VALUES (?1)",
&[SqliteValue::Text("never".to_string())],
)
.await
{
Outcome::Cancelled(_) => {}
other => panic!("expected cancellation, got: {other:?}"),
}
let rows = match conn.query(&cx, "SELECT COUNT(*) FROM t", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("count query failed: {other:?}"),
};
assert_eq!(rows[0].get_idx(0).unwrap().as_integer(), Some(0));
});
}
#[cfg(feature = "sqlite")]
mod pragma_journal_mode_conformance {
use super::*;
use crate::test_utils::run_test_with_cx;
use std::fs;
use std::path::PathBuf;
use tempfile::TempDir;
struct JournalModeTestData {
temp_dir: TempDir,
db_path: PathBuf,
}
impl JournalModeTestData {
fn new() -> Self {
let temp_dir = tempfile::tempdir().expect("Failed to create temp directory");
let db_path = temp_dir.path().join("test.db");
Self { temp_dir, db_path }
}
fn get_db_path(&self) -> &Path {
&self.db_path
}
fn get_wal_path(&self) -> PathBuf {
self.db_path.with_extension("db-wal")
}
fn get_shm_path(&self) -> PathBuf {
self.db_path.with_extension("db-shm")
}
async fn get_journal_mode(conn: &SqliteConnection, cx: &Cx) -> String {
let rows = match conn.query_unchecked(cx, "PRAGMA journal_mode", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("Failed to query journal_mode: {other:?}"),
};
rows[0]
.get_idx(0)
.unwrap()
.as_text()
.unwrap_or_else(|| panic!("journal_mode should return a string"))
.to_owned()
}
async fn set_journal_mode(
conn: &SqliteConnection,
cx: &Cx,
mode: &str,
) -> Outcome<String, SqliteError> {
let sql = format!("PRAGMA journal_mode = {}", mode);
match conn.query_unchecked(cx, &sql, &[]).await {
Outcome::Ok(rows) => Outcome::Ok(
rows[0]
.get_idx(0)
.unwrap()
.as_text()
.unwrap_or_else(|| panic!("journal_mode pragma should return a string"))
.to_owned(),
),
Outcome::Err(err) => Outcome::Err(err),
Outcome::Cancelled(cancelled) => Outcome::Cancelled(cancelled),
Outcome::Panicked(payload) => Outcome::Panicked(payload),
}
}
async fn setup_test_data(conn: &SqliteConnection, cx: &Cx) {
match conn
.execute_batch(
cx,
"
CREATE TABLE test_data (
id INTEGER PRIMARY KEY,
value TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO test_data (value) VALUES ('test1'), ('test2'), ('test3');
",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("Failed to create test data: {other:?}"),
}
}
async fn verify_test_data(conn: &SqliteConnection, cx: &Cx, expected_count: i64) {
let rows = match conn.query(cx, "SELECT COUNT(*) FROM test_data", &[]).await {
Outcome::Ok(rows) => rows,
other => panic!("Failed to count test data: {other:?}"),
};
let count = rows[0].get_idx(0).unwrap().as_integer().unwrap();
assert_eq!(count, expected_count, "Test data count mismatch");
}
}
#[test]
fn delete_to_wal_mode_transition_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
let initial_mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
assert_eq!(
initial_mode.to_lowercase(),
"wal",
"asupersync connection defaults should enable WAL"
);
let delete_result =
match JournalModeTestData::set_journal_mode(&conn, &cx, "DELETE").await {
Outcome::Ok(mode) => mode,
other => panic!("Failed to set DELETE mode: {other:?}"),
};
assert_eq!(
delete_result.to_lowercase(),
"delete",
"Should start in DELETE mode"
);
JournalModeTestData::setup_test_data(&conn, &cx).await;
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
let wal_result =
match JournalModeTestData::set_journal_mode(&conn, &cx, "WAL").await {
Outcome::Ok(mode) => mode,
other => panic!("Failed to set WAL mode: {other:?}"),
};
assert_eq!(
wal_result.to_lowercase(),
"wal",
"Should transition to WAL mode"
);
let current_mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
assert_eq!(
current_mode.to_lowercase(),
"wal",
"Journal mode should be WAL"
);
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
match conn
.execute(
&cx,
"INSERT INTO test_data (value) VALUES (?)",
&[SqliteValue::Text("wal_data".to_owned())],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to insert WAL data: {other:?}"),
};
assert!(
test_data.get_wal_path().exists(),
"WAL file should be created"
);
assert!(
test_data.get_shm_path().exists(),
"SHM file should be created"
);
JournalModeTestData::verify_test_data(&conn, &cx, 4).await;
conn.close().unwrap();
});
}
#[test]
fn wal_to_truncate_mode_transition_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
match JournalModeTestData::set_journal_mode(&conn, &cx, "WAL").await {
Outcome::Ok(_) => {}
other => panic!("Failed to set WAL mode: {other:?}"),
};
JournalModeTestData::setup_test_data(&conn, &cx).await;
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
assert!(test_data.get_wal_path().exists(), "WAL file should exist");
let truncate_result =
match JournalModeTestData::set_journal_mode(&conn, &cx, "TRUNCATE").await {
Outcome::Ok(mode) => mode,
other => panic!("Failed to set TRUNCATE mode: {other:?}"),
};
assert_eq!(
truncate_result.to_lowercase(),
"truncate",
"Should transition to TRUNCATE mode"
);
let current_mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
assert_eq!(
current_mode.to_lowercase(),
"truncate",
"Journal mode should be TRUNCATE"
);
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
match conn
.execute(
&cx,
"INSERT INTO test_data (value) VALUES (?)",
&[SqliteValue::Text("truncate_data".to_owned())],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to insert TRUNCATE data: {other:?}"),
};
JournalModeTestData::verify_test_data(&conn, &cx, 4).await;
conn.close().unwrap();
});
}
#[test]
fn memory_mode_persistence_loss_conformance() {
run_test_with_cx(|cx| async move {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open in-memory connection: {other:?}"),
};
let memory_result =
match JournalModeTestData::set_journal_mode(&conn, &cx, "MEMORY").await {
Outcome::Ok(mode) => mode,
other => panic!("Failed to set MEMORY mode: {other:?}"),
};
assert_eq!(
memory_result.to_lowercase(),
"memory",
"Should be in MEMORY mode"
);
JournalModeTestData::setup_test_data(&conn, &cx).await;
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
match conn
.execute_batch_unchecked(
&cx,
"
BEGIN TRANSACTION;
INSERT INTO test_data (value) VALUES ('memory_test');
UPDATE test_data SET value = 'modified' WHERE id = 1;
",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("Failed to begin transaction: {other:?}"),
};
drop(conn);
let new_conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to reopen in-memory connection: {other:?}"),
};
let tables_result = new_conn
.query(
&cx,
"SELECT name FROM sqlite_master WHERE type='table'",
&[],
)
.await;
match tables_result {
Outcome::Ok(rows) => {
assert_eq!(
rows.len(),
0,
"In-memory database should have no persistent tables"
);
}
other => panic!("Failed to query sqlite_master: {other:?}"),
}
new_conn.close().unwrap();
});
}
#[test]
fn off_mode_atomicity_absence_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
let off_result =
match JournalModeTestData::set_journal_mode(&conn, &cx, "OFF").await {
Outcome::Ok(mode) => mode,
other => panic!("Failed to set OFF mode: {other:?}"),
};
assert_eq!(off_result.to_lowercase(), "off", "Should be in OFF mode");
match conn
.execute_batch(
&cx,
"
CREATE TABLE atomicity_test (
id INTEGER PRIMARY KEY,
step INTEGER,
data TEXT
);
",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("Failed to create table: {other:?}"),
};
match conn.execute_unchecked(&cx, "BEGIN TRANSACTION", &[]).await {
Outcome::Ok(_) => {}
other => panic!("Failed to begin transaction: {other:?}"),
};
match conn
.execute(
&cx,
"INSERT INTO atomicity_test (step, data) VALUES (1, 'step1')",
&[],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to insert step1: {other:?}"),
};
match conn
.execute(
&cx,
"INSERT INTO atomicity_test (step, data) VALUES (2, 'step2')",
&[],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to insert step2: {other:?}"),
};
match conn.execute_unchecked(&cx, "COMMIT", &[]).await {
Outcome::Ok(_) => {}
other => panic!("Failed to commit: {other:?}"),
};
let rows = match conn
.query(&cx, "SELECT COUNT(*) FROM atomicity_test", &[])
.await
{
Outcome::Ok(rows) => rows,
other => panic!("Failed to count rows: {other:?}"),
};
let count = rows[0].get_idx(0).unwrap().as_integer().unwrap();
assert_eq!(count, 2, "Both inserts should be present");
let journal_files = fs::read_dir(test_data.temp_dir.path())
.unwrap()
.filter_map(|entry| entry.ok())
.filter(|entry| {
entry
.path()
.extension()
.is_some_and(|ext| ext == "journal" || ext == "wal" || ext == "shm")
})
.count();
assert_eq!(journal_files, 0, "OFF mode should not create journal files");
conn.close().unwrap();
});
}
#[test]
fn unsupported_mode_fallback_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
let invalid_modes = ["INVALID", "BOGUS", "NONEXISTENT"];
for invalid_mode in &invalid_modes {
match JournalModeTestData::set_journal_mode(&conn, &cx, invalid_mode).await {
Outcome::Ok(returned_mode) => {
assert_ne!(
returned_mode.to_lowercase(),
invalid_mode.to_lowercase(),
"Should not accept invalid mode: {}",
invalid_mode
);
let valid_modes =
["delete", "truncate", "persist", "memory", "wal", "off"];
assert!(
valid_modes.contains(&returned_mode.to_lowercase().as_str()),
"Fallback should be a valid journal mode, got: {}",
returned_mode
);
}
Outcome::Err(_) => {
}
other => panic!(
"Unexpected outcome for invalid mode {}: {other:?}",
invalid_mode
),
}
let current_mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
assert!(
!current_mode.is_empty(),
"Should still have a valid journal mode after invalid attempt"
);
}
JournalModeTestData::setup_test_data(&conn, &cx).await;
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
conn.close().unwrap();
});
}
#[test]
fn journal_mode_persistence_across_connections_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
{
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
match JournalModeTestData::set_journal_mode(&conn, &cx, "WAL").await {
Outcome::Ok(_) => {}
other => panic!("Failed to set WAL mode: {other:?}"),
};
JournalModeTestData::setup_test_data(&conn, &cx).await;
conn.close().unwrap();
}
{
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to reopen connection: {other:?}"),
};
let persistent_mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
assert_eq!(
persistent_mode.to_lowercase(),
"wal",
"WAL mode should persist across connections"
);
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
conn.close().unwrap();
}
});
}
#[test]
fn journal_mode_concurrent_access_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
match JournalModeTestData::set_journal_mode(&conn, &cx, "WAL").await {
Outcome::Ok(_) => {}
other => panic!("Failed to set WAL mode: {other:?}"),
};
JournalModeTestData::setup_test_data(&conn, &cx).await;
let reader_conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open reader connection: {other:?}"),
};
JournalModeTestData::verify_test_data(&conn, &cx, 3).await;
JournalModeTestData::verify_test_data(&reader_conn, &cx, 3).await;
match conn
.execute(
&cx,
"INSERT INTO test_data (value) VALUES (?)",
&[SqliteValue::Text("concurrent_write".to_owned())],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed concurrent write: {other:?}"),
};
JournalModeTestData::verify_test_data(&conn, &cx, 4).await;
reader_conn.close().unwrap();
conn.close().unwrap();
});
}
#[test]
fn journal_mode_edge_cases_conformance() {
run_test_with_cx(|cx| async move {
let test_data = JournalModeTestData::new();
let conn = match SqliteConnection::open(&cx, test_data.get_db_path()).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
let modes_to_test = [
("wal", "wal"),
("WAL", "wal"),
("Wal", "wal"),
("DELETE", "delete"),
("delete", "delete"),
];
for (input_mode, expected_mode) in &modes_to_test {
match JournalModeTestData::set_journal_mode(&conn, &cx, input_mode).await {
Outcome::Ok(returned_mode) => {
assert_eq!(
returned_mode.to_lowercase(),
expected_mode.to_lowercase(),
"Mode {} should normalize to {}",
input_mode,
expected_mode
);
}
other => panic!("Failed to set mode {}: {other:?}", input_mode),
}
}
for _ in 0..5 {
let mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
assert!(
!mode.is_empty(),
"Journal mode query should always return a value"
);
}
let current_mode = JournalModeTestData::get_journal_mode(&conn, &cx).await;
match JournalModeTestData::set_journal_mode(&conn, &cx, ¤t_mode).await {
Outcome::Ok(returned_mode) => {
assert_eq!(
returned_mode.to_lowercase(),
current_mode.to_lowercase(),
"Setting to current mode should be no-op"
);
}
other => panic!("Failed to set to current mode: {other:?}"),
}
conn.close().unwrap();
});
}
}
mod real_database_integration {
use super::*;
use crate::test_utils::run_test_with_cx;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Instant;
struct RealSqliteConfig {
database_path: String,
enabled: bool,
reason: Option<String>,
}
impl RealSqliteConfig {
fn new() -> Self {
let enabled = std::env::var("REAL_SQLITE_TESTS").unwrap_or_default() == "true";
let db_path =
std::env::var("SQLITE_TEST_PATH").unwrap_or_else(|_| ":memory:".to_string());
let reason = if !enabled {
Some("REAL_SQLITE_TESTS not set to 'true'".to_string())
} else if std::env::var("NODE_ENV").unwrap_or_default() == "production" {
Some("BLOCKED: NODE_ENV=production".to_string())
} else if db_path.contains("prod") || db_path.contains("/var/lib/") {
Some("BLOCKED: Production database path detected".to_string())
} else {
None
};
Self {
database_path: db_path,
enabled: enabled && reason.is_none(),
reason,
}
}
}
#[derive(Debug)]
struct SqliteTestLogger {
test_name: String,
start_time: Instant,
phase_count: AtomicU32,
}
impl SqliteTestLogger {
fn new(test_name: &str) -> Self {
let logger = Self {
test_name: test_name.to_string(),
start_time: Instant::now(),
phase_count: AtomicU32::new(0),
};
eprintln!(
"{{\"test\":\"{}\",\"event\":\"test_start\",\"ts\":\"{}\"}}",
test_name,
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
);
logger
}
fn phase(&self, phase_name: &str) {
let phase_num = self.phase_count.fetch_add(1, Ordering::Relaxed);
let elapsed_ms = self.start_time.elapsed().as_millis();
eprintln!(
"{{\"test\":\"{}\",\"event\":\"phase\",\"phase\":\"{}\",\"phase_num\":{},\"elapsed_ms\":{},\"ts\":{}}}",
self.test_name,
phase_name,
phase_num,
elapsed_ms,
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
);
}
fn sqlite_operation(&self, operation: &str, result: &str, details: Option<&str>) {
let mut log_entry = format!(
"{{\"test\":\"{}\",\"event\":\"sqlite_operation\",\"operation\":\"{}\",\"result\":\"{}\"",
self.test_name, operation, result
);
if let Some(detail) = details {
log_entry.push_str(&format!(",\"details\":\"{}\"", detail));
}
log_entry.push_str(&format!(
",\"ts\":{}}}",
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
));
eprintln!("{}", log_entry);
}
fn assert_match(&self, field: &str, expected: &str, actual: &str) -> bool {
let matches = expected == actual;
eprintln!(
"{{\"test\":\"{}\",\"event\":\"assertion\",\"field\":\"{}\",\"expected\":\"{}\",\"actual\":\"{}\",\"matches\":{},\"ts\":{}}}",
self.test_name,
field,
expected,
actual,
matches,
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
);
matches
}
fn test_end(&self, result: &str) {
let duration_ms = self.start_time.elapsed().as_millis();
eprintln!(
"{{\"test\":\"{}\",\"event\":\"test_end\",\"result\":\"{}\",\"duration_ms\":{},\"ts\":{}}}",
self.test_name,
result,
duration_ms,
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
);
}
}
struct SqliteDataFactory {
counter: AtomicU32,
}
impl SqliteDataFactory {
fn new() -> Self {
Self {
counter: AtomicU32::new(0),
}
}
fn create_user_record(&self) -> (i64, String, String) {
let id = self.counter.fetch_add(1, Ordering::Relaxed) as i64;
let name = format!("user_{}", id);
let email = format!("user{}@test-domain.com", id);
(id, name, email)
}
fn create_batch_records(&self, count: usize) -> Vec<(String, String, i64)> {
(0..count)
.map(|_| {
let (id, name, email) = self.create_user_record();
(name, email, id)
})
.collect()
}
fn create_transaction_batch(
&self,
user_id: i64,
count: usize,
) -> Vec<(i64, String, f64)> {
(0..count)
.map(|i| {
let tx_id = self.counter.fetch_add(1, Ordering::Relaxed) as i64;
let description = format!("Transaction {} for user {}", i, user_id);
let amount = (i as f64) * 10.5 + 1.0; (tx_id, description, amount)
})
.collect()
}
}
fn require_real_sqlite() -> Option<RealSqliteConfig> {
let config = RealSqliteConfig::new();
if !config.enabled {
let reason = config
.reason
.as_deref()
.unwrap_or("Real SQLite testing not available");
eprintln!("SKIPPING: {}", reason);
return None;
}
Some(config)
}
#[test]
fn test_real_sqlite_journal_mode_transitions() {
let Some(config) = require_real_sqlite() else {
return;
};
let log = SqliteTestLogger::new("real_sqlite_journal_mode_transitions");
run_test_with_cx(|cx| async move {
log.phase("setup");
let conn = if config.database_path == ":memory:" {
match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open in-memory connection: {other:?}"),
}
} else {
match SqliteConnection::open(&cx, &config.database_path).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open file connection: {other:?}"),
}
};
log.phase("transaction_isolation_setup");
match conn.execute_unchecked(&cx, "BEGIN TRANSACTION", &[]).await {
Outcome::Ok(_) => log.sqlite_operation("begin_transaction", "success", None),
other => panic!("Failed to begin transaction: {other:?}"),
}
log.phase("schema_and_data_setup");
let factory = SqliteDataFactory::new();
match conn
.execute_batch(
&cx,
"
CREATE TABLE users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE transactions (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
description TEXT NOT NULL,
amount REAL NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id)
);
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_transactions_user_id ON transactions(user_id);
",
)
.await
{
Outcome::Ok(()) => log.sqlite_operation("schema_creation", "success", None),
other => panic!("Failed to create schema: {other:?}"),
}
let users = factory.create_batch_records(10);
for (name, email, user_id) in &users {
match conn
.execute(
&cx,
"INSERT INTO users (id, name, email) VALUES (?1, ?2, ?3)",
&[
SqliteValue::Integer(*user_id),
SqliteValue::Text(name.clone()),
SqliteValue::Text(email.clone()),
],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to insert user: {other:?}"),
}
let transactions = factory.create_transaction_batch(*user_id, 3);
for (tx_id, description, amount) in transactions {
match conn
.execute(
&cx,
"INSERT INTO transactions (id, user_id, description, amount) VALUES (?1, ?2, ?3, ?4)",
&[
SqliteValue::Integer(tx_id),
SqliteValue::Integer(*user_id),
SqliteValue::Text(description),
SqliteValue::Real(amount),
],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to insert transaction: {other:?}"),
}
}
}
log.sqlite_operation(
"test_data_inserted",
"success",
Some(&format!(
"{} users, {} transactions",
users.len(),
users.len().saturating_mul(3)
)),
);
log.phase("journal_mode_testing");
let initial_mode = match conn.query_unchecked(&cx, "PRAGMA journal_mode", &[]).await
{
Outcome::Ok(rows) => rows[0].get_idx(0).unwrap().as_text().unwrap().to_owned(),
other => panic!("Failed to get initial journal mode: {other:?}"),
};
log.sqlite_operation("get_initial_journal_mode", "success", Some(&initial_mode));
let user_count_before =
match conn.query(&cx, "SELECT COUNT(*) FROM users", &[]).await {
Outcome::Ok(rows) => rows[0].get_idx(0).unwrap().as_integer().unwrap(),
other => panic!("Failed to count users: {other:?}"),
};
assert!(log.assert_match(
"user_count_before_journal_change",
"10",
&user_count_before.to_string()
));
log.phase("wal_mode_transition");
match conn
.query_unchecked(&cx, "PRAGMA journal_mode = WAL", &[])
.await
{
Outcome::Ok(rows) => {
let new_mode = rows[0].get_idx(0).unwrap().as_text().unwrap();
log.sqlite_operation("set_journal_mode_wal", "success", Some(new_mode));
if config.database_path != ":memory:" {
assert!(log.assert_match(
"journal_mode_after_wal",
"wal",
&new_mode.to_lowercase()
));
}
}
other => panic!("Failed to set WAL mode: {other:?}"),
}
log.phase("data_integrity_verification");
let user_count_after =
match conn.query(&cx, "SELECT COUNT(*) FROM users", &[]).await {
Outcome::Ok(rows) => rows[0].get_idx(0).unwrap().as_integer().unwrap(),
other => panic!("Failed to count users after mode change: {other:?}"),
};
assert!(log.assert_match(
"user_count_after_journal_change",
"10",
&user_count_after.to_string()
));
let tx_count = match conn
.query(&cx, "SELECT COUNT(*) FROM transactions", &[])
.await
{
Outcome::Ok(rows) => rows[0].get_idx(0).unwrap().as_integer().unwrap(),
other => panic!("Failed to count transactions: {other:?}"),
};
assert!(log.assert_match("transaction_count", "30", &tx_count.to_string()));
log.phase("complex_query_testing");
let user_tx_summary = match conn
.query(
&cx,
"SELECT u.name, COUNT(t.id) as tx_count, SUM(t.amount) as total_amount
FROM users u
LEFT JOIN transactions t ON u.id = t.user_id
GROUP BY u.id, u.name
ORDER BY total_amount DESC
LIMIT 5",
&[],
)
.await
{
Outcome::Ok(rows) => rows,
other => panic!("Failed to execute complex query: {other:?}"),
};
assert!(
user_tx_summary.len() >= 5,
"Should have at least 5 users in summary"
);
log.sqlite_operation(
"complex_query",
"success",
Some(&format!("{} user summaries", user_tx_summary.len())),
);
log.phase("transaction_rollback");
match conn.execute_unchecked(&cx, "ROLLBACK", &[]).await {
Outcome::Ok(_) => log.sqlite_operation("rollback_transaction", "success", None),
other => panic!("Failed to rollback transaction: {other:?}"),
}
log.phase("cleanup");
conn.close().unwrap();
log.test_end("pass");
});
}
#[test]
fn test_real_sqlite_concurrent_access_patterns() {
let Some(_config) = require_real_sqlite() else {
return;
};
let log = SqliteTestLogger::new("real_sqlite_concurrent_access");
run_test_with_cx(|cx| async move {
log.phase("setup");
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("Failed to open connection: {other:?}"),
};
log.phase("wal_mode_setup");
match conn
.query_unchecked(&cx, "PRAGMA journal_mode = WAL", &[])
.await
{
Outcome::Ok(_) => log.sqlite_operation("set_wal_mode", "success", None),
other => panic!("Failed to set WAL mode: {other:?}"),
}
log.phase("schema_setup");
match conn.execute_unchecked(&cx, "BEGIN TRANSACTION", &[]).await {
Outcome::Ok(_) => {}
other => panic!("Failed to begin transaction: {other:?}"),
}
match conn
.execute_batch(
&cx,
"
CREATE TABLE accounts (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
balance REAL NOT NULL DEFAULT 0.0,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE transfers (
id INTEGER PRIMARY KEY,
from_account INTEGER NOT NULL,
to_account INTEGER NOT NULL,
amount REAL NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (from_account) REFERENCES accounts(id),
FOREIGN KEY (to_account) REFERENCES accounts(id)
);
",
)
.await
{
Outcome::Ok(()) => log.sqlite_operation("concurrent_schema", "success", None),
other => panic!("Failed to create concurrent test schema: {other:?}"),
}
log.phase("test_data_creation");
let accounts = vec![
(1, "Account A", 1000.0),
(2, "Account B", 500.0),
(3, "Account C", 750.0),
];
for (id, name, balance) in &accounts {
match conn
.execute(
&cx,
"INSERT INTO accounts (id, name, balance) VALUES (?1, ?2, ?3)",
&[
SqliteValue::Integer(*id),
SqliteValue::Text(name.to_string()),
SqliteValue::Real(*balance),
],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to create account: {other:?}"),
}
}
log.phase("concurrent_operations_simulation");
let transfers = vec![
(1, 2, 100.0), (2, 3, 200.0), (3, 1, 150.0), ];
for (from_id, to_id, amount) in &transfers {
let balance_check = match conn
.query(
&cx,
"SELECT balance FROM accounts WHERE id = ?1",
&[SqliteValue::Integer(*from_id)],
)
.await
{
Outcome::Ok(rows) => rows[0].get_idx(0).unwrap().as_real().unwrap(),
other => panic!("Failed to check balance: {other:?}"),
};
if balance_check >= *amount {
match conn
.execute(
&cx,
"INSERT INTO transfers (from_account, to_account, amount, status) VALUES (?1, ?2, ?3, 'completed')",
&[
SqliteValue::Integer(*from_id),
SqliteValue::Integer(*to_id),
SqliteValue::Real(*amount),
],
)
.await
{
Outcome::Ok(_) => log.sqlite_operation("transfer_created", "success", Some(&format!("{} -> {}: {}", from_id, to_id, amount))),
other => panic!("Failed to create transfer: {other:?}"),
}
match conn
.execute(
&cx,
"UPDATE accounts SET balance = balance - ?1 WHERE id = ?2",
&[SqliteValue::Real(*amount), SqliteValue::Integer(*from_id)],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to debit account: {other:?}"),
}
match conn
.execute(
&cx,
"UPDATE accounts SET balance = balance + ?1 WHERE id = ?2",
&[SqliteValue::Real(*amount), SqliteValue::Integer(*to_id)],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("Failed to credit account: {other:?}"),
}
}
}
log.phase("integrity_verification");
let final_balances = match conn
.query(
&cx,
"SELECT id, name, balance FROM accounts ORDER BY id",
&[],
)
.await
{
Outcome::Ok(rows) => rows,
other => panic!("Failed to get final balances: {other:?}"),
};
for row in &final_balances {
let id = row.get_idx(0).unwrap().as_integer().unwrap();
let name = row.get_idx(1).unwrap().as_text().unwrap();
let balance = row.get_idx(2).unwrap().as_real().unwrap();
log.sqlite_operation(
"final_balance",
"verified",
Some(&format!("{} ({}): {}", name, id, balance)),
);
}
let transfer_count = match conn
.query(
&cx,
"SELECT COUNT(*) FROM transfers WHERE status = 'completed'",
&[],
)
.await
{
Outcome::Ok(rows) => rows[0].get_idx(0).unwrap().as_integer().unwrap(),
other => panic!("Failed to count transfers: {other:?}"),
};
assert!(transfer_count > 0, "Should have completed transfers");
log.sqlite_operation(
"transfer_verification",
"success",
Some(&format!("{} completed transfers", transfer_count)),
);
log.phase("rollback_cleanup");
match conn.execute_unchecked(&cx, "ROLLBACK", &[]).await {
Outcome::Ok(_) => log.sqlite_operation("rollback", "success", None),
other => panic!("Failed to rollback: {other:?}"),
}
conn.close().unwrap();
log.test_end("pass");
});
}
}
mod sqlite_prepared_statement_reset_audit {
use super::*;
#[test]
fn audit_rusqlite_automatic_statement_reset() {
init_test_logging();
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE reset_test (id INTEGER PRIMARY KEY, value TEXT);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
for i in 1..=5 {
let value = format!("test-value-{i}");
match conn
.execute(
&cx,
"INSERT INTO reset_test (value) VALUES (?1)",
&[SqliteValue::Text(value)],
)
.await
{
Outcome::Ok(rows) => {
crate::assert_with_log!(
rows == 1,
"INSERT should affect exactly 1 row",
1,
rows
);
}
other => panic!("insert {i} failed: {other:?}"),
}
}
for i in 1..=5 {
let expected_value = format!("test-value-{i}");
match conn
.query(
&cx,
"SELECT value FROM reset_test WHERE id = ?1",
&[SqliteValue::Integer(i)],
)
.await
{
Outcome::Ok(rows) => {
crate::assert_with_log!(
rows.len() == 1,
"Query should return exactly 1 row",
1,
rows.len()
);
let actual_value = rows[0].get_str("value").unwrap();
crate::assert_with_log!(
actual_value == expected_value,
"Query result should match inserted value",
&expected_value,
actual_value
);
}
other => panic!("query {i} failed: {other:?}"),
}
}
eprintln!(
"{{\"audit\":\"SQLITE_RESET_SEMANTICS\",\"status\":\"SOUND\",\"requirement\":\"automatic statement reset via rusqlite APIs\"}}"
);
crate::test_complete!("audit_rusqlite_automatic_statement_reset");
});
}
#[test]
fn audit_prepare_cached_statement_reuse() {
init_test_logging();
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE cached_test (id INTEGER PRIMARY KEY, data TEXT);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
{
let guard = conn.inner.lock();
let raw_conn = guard.get().expect("connection should be open");
raw_conn.set_prepared_statement_cache_capacity(2);
}
match conn
.execute(&cx, "INSERT INTO cached_test (data) VALUES ('first')", &[])
.await
{
Outcome::Ok(_) => {}
other => panic!("insert first failed: {other:?}"),
}
const QUERY_SQL: &str = "SELECT data FROM cached_test WHERE id = ?1";
match conn.query(&cx, QUERY_SQL, &[SqliteValue::Integer(1)]).await {
Outcome::Ok(rows) => {
crate::assert_with_log!(
rows.len() == 1 && rows[0].get_str("data").unwrap() == "first",
"First query execution should return 'first'",
"first",
rows[0].get_str("data").unwrap()
);
}
other => panic!("first query failed: {other:?}"),
}
match conn.query(&cx, QUERY_SQL, &[SqliteValue::Integer(1)]).await {
Outcome::Ok(rows) => {
crate::assert_with_log!(
rows.len() == 1 && rows[0].get_str("data").unwrap() == "first",
"Second query execution should return same result",
"first",
rows[0].get_str("data").unwrap()
);
}
other => panic!("second query failed: {other:?}"),
}
match conn
.execute(&cx, "INSERT INTO cached_test (data) VALUES ('second')", &[])
.await
{
Outcome::Ok(_) => {}
other => panic!("insert second failed: {other:?}"),
}
match conn.query(&cx, QUERY_SQL, &[SqliteValue::Integer(2)]).await {
Outcome::Ok(rows) => {
crate::assert_with_log!(
rows.len() == 1 && rows[0].get_str("data").unwrap() == "second",
"Cached statement with new parameter should return correct result",
"second",
rows[0].get_str("data").unwrap()
);
}
other => panic!("parameter change query failed: {other:?}"),
}
eprintln!(
"{{\"audit\":\"STATEMENT_CACHE_RESET\",\"status\":\"SOUND\",\"requirement\":\"cached statement reset between executions\"}}"
);
crate::test_complete!("audit_prepare_cached_statement_reuse");
});
}
#[test]
fn audit_query_iterator_reset_on_drop() {
init_test_logging();
let cx = create_test_cx();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
match conn
.execute_batch(
&cx,
"CREATE TABLE iterator_test (id INTEGER PRIMARY KEY, value INTEGER);",
)
.await
{
Outcome::Ok(()) => {}
other => panic!("create table failed: {other:?}"),
}
for i in 1..=10 {
match conn
.execute(
&cx,
"INSERT INTO iterator_test (value) VALUES (?1)",
&[SqliteValue::Integer(i * 10)],
)
.await
{
Outcome::Ok(_) => {}
other => panic!("insert {i} failed: {other:?}"),
}
}
let query_sql = "SELECT COUNT(*) as count FROM iterator_test WHERE value > ?1";
let count_gt_0 = match conn
.query_row(&cx, query_sql, &[SqliteValue::Integer(0)])
.await
{
Outcome::Ok(Some(row)) => row.get_i64("count").unwrap(),
other => panic!("count_gt_0 query failed: {other:?}"),
};
let count_gt_50 = match conn
.query_row(&cx, query_sql, &[SqliteValue::Integer(50)])
.await
{
Outcome::Ok(Some(row)) => row.get_i64("count").unwrap(),
other => panic!("count_gt_50 query failed: {other:?}"),
};
let count_gt_100 = match conn
.query_row(&cx, query_sql, &[SqliteValue::Integer(100)])
.await
{
Outcome::Ok(Some(row)) => row.get_i64("count").unwrap(),
other => panic!("count_gt_100 query failed: {other:?}"),
};
crate::assert_with_log!(
count_gt_0 == 10 && count_gt_50 == 5 && count_gt_100 == 0,
"Statement reset between queries should produce correct results",
(10, 5, 0),
(count_gt_0, count_gt_50, count_gt_100)
);
eprintln!(
"{{\"audit\":\"ITERATOR_DROP_RESET\",\"status\":\"SOUND\",\"requirement\":\"statement reset on Rows drop\"}}"
);
crate::test_complete!("audit_query_iterator_reset_on_drop");
});
}
#[test]
fn audit_sqlite_query_result_streaming_memory_usage() {
eprintln!(
"{{\"defect\":\"SQLITE_QUERY_RESULT_STREAMING\",\"severity\":\"CRITICAL\",\"impact\":\"OOM risk\",\"violation\":\"sqlite3_step streaming\",\"architecture\":\"blocking_pool\",\"complexity\":\"HIGH\"}}"
);
eprintln!(
"{{\"recommendation\":\"FILE_BEAD\",\"reason\":\"30min_deadline_insufficient\",\"estimated_effort\":\"2-4_hours\",\"same_pattern_as\":\"MySQL/PostgreSQL but blocking_pool_architecture\"}}"
);
}
}
#[test]
fn reserve_transaction_obligation_skips_root_region() {
let non_root = Cx::for_testing();
let token = reserve_transaction_obligation(&non_root);
assert!(
token.is_some(),
"non-root transaction must be obligation-tracked"
);
if let Some(token) = token {
let _ = token.abort();
}
let root = Cx::new(
RegionId::from_arena(ArenaIndex::new(0, 0)),
TaskId::from_arena(ArenaIndex::new(0, 0)),
Budget::INFINITE,
);
assert!(
reserve_transaction_obligation(&root).is_none(),
"root-region transaction is not obligation-tracked (ASUP-E103)"
);
}
#[test]
fn dropped_transaction_with_obligation_aborts_and_poisons() {
let cx = Cx::for_testing();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
{
let tx = match conn.begin(&cx).await {
Outcome::Ok(tx) => tx,
Outcome::Err(e) => panic!("begin failed: {e}"),
Outcome::Cancelled(r) => panic!("begin cancelled: {r:?}"),
Outcome::Panicked(p) => panic!("begin panicked: {p:?}"),
};
assert!(
tx.obligation.is_some(),
"for_testing cx is non-root, so the obligation must be reserved"
);
}
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::NeedsRollback,
"dropped transaction must poison the connection for rollback"
);
});
}
#[test]
fn committed_transaction_discharges_obligation_without_leak() {
let cx = Cx::for_testing();
block_on(async {
let conn = match SqliteConnection::open_in_memory(&cx).await {
Outcome::Ok(conn) => conn,
other => panic!("open_in_memory failed: {other:?}"),
};
let tx = match conn.begin(&cx).await {
Outcome::Ok(tx) => tx,
Outcome::Err(e) => panic!("begin failed: {e}"),
Outcome::Cancelled(r) => panic!("begin cancelled: {r:?}"),
Outcome::Panicked(p) => panic!("begin panicked: {p:?}"),
};
match tx.commit(&cx).await {
Outcome::Ok(()) => {}
other => panic!("commit failed: {other:?}"),
}
assert_eq!(
*conn.transaction_state.lock(),
TransactionState::Autocommit,
"committed transaction must return to autocommit"
);
});
}
}