Replace per-connection execution lock with statement-level interrupts

Query timeouts previously relied on a per-connection execution lock that
serialized every operation so the timer wheel's connection.interrupt() would
unambiguously hit the timed-out operation. This serialized all concurrent work
on a connection and added a mutex acquisition to every call.

Use libsql_stmt_interrupt() instead: the timer wheel now interrupts the
specific statement that timed out (via a new Interruptible trait, so the wheel
can target either a statement or a connection), leaving other concurrent
operations on the same connection untouched. This removes execution_lock
entirely, along with acquire_execution_lock and the deadline/remaining
plumbing it required.

Requires libsql 0.10.0-pre.4, which honors libsql_stmt_interrupt per VDBE
instruction (pre.3 only checked at step entry and could not abort a long
single step such as an aggregate).

The per-statement interrupt flag is sticky: sqlite3_step checks it at entry and
only reset() clears it, unlike the connection flag SQLite auto-clears each
step. We do not reset before every step (that would add overhead to every
call); each operation recovers at its existing natural reset point instead -
run/iterate reset before stepping, get/get_sync reset after (self-healing).

Add tests: statement-level interrupt of a multi-row query and a single-step
aggregate, the sticky-flag-needs-reset contract, and a JS test for statement
recovery after an interrupt.

Author: penberg

Cargo.toml

@@ -11,7 +11,7 @@ crate-type = ["cdylib"]
 
 [dependencies]
 glob-match = "0.2"
-libsql = { version = "0.10.0-pre.3", features = ["encryption"]  }
+libsql = { version = "0.10.0-pre.4", features = ["encryption"]  }
 napi = { version = "2", default-features = false, features = ["napi6", "tokio_rt", "async"] }
 napi-derive = "2"
 once_cell = "1.18.0"

integration-tests/tests/async.test.js

@@ -548,7 +548,7 @@ test.serial("Timeout on Statement.get() does not leak into later prepare()/EXPLA
   db.close();
 });
 
-test.serial("Query timeout covers wait time on the execution lock", async (t) => {
+test.serial("Query timeout interrupts long-running queries under concurrency", async (t) => {
   t.timeout(15_000);
   const [db, errorType] = await connect(":memory:", { defaultQueryTimeout: 200 });
   await db.exec("CREATE TABLE t(x INTEGER)");

src/lib.rs

@@ -37,7 +37,7 @@ use std::{
         atomic::{AtomicBool, Ordering},
         Arc, Mutex,
     },
-    time::{Duration, Instant},
+    time::Duration,
 };
 use tokio::runtime::Runtime;
 use tracing_subscriber::{filter::LevelFilter, EnvFilter};
@@ -238,8 +238,6 @@ pub struct Database {
     memory: bool,
     // Maximum time in milliseconds that a query is allowed to run.
     query_timeout: Option<Duration>,
-    // Ensures only one operation executes per connection at a time.
-    execution_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 impl Drop for Database {
@@ -341,14 +339,12 @@ pub async fn connect(path: String, opts: Option<Options>) -> Result<Database> {
         .as_ref()
         .and_then(|o| o.defaultQueryTimeout)
         .and_then(query_timeout_duration);
-    let execution_lock = Arc::new(tokio::sync::Mutex::new(()));
     Ok(Database {
         db: Some(db),
         conn: Some(conn),
         default_safe_integers,
         memory,
         query_timeout,
-        execution_lock,
     })
 }
 
@@ -404,17 +400,15 @@ impl Database {
                 ));
             }
         };
-        let (_execution_guard, deadline) =
-            acquire_execution_lock(&self.execution_lock, self.query_timeout).await?;
-        let timeout_guard = register_remaining_timeout(&conn, deadline)?;
+        let timeout_guard = register_timeout(&conn, self.query_timeout);
         let stmt = match conn.prepare(&sql).await {
             Ok(stmt) => stmt,
             Err(err) if is_sqlite_interrupt(&err) => {
                 // Drop our guard before clear_stale_interrupt so the bg thread
                 // can't fire conn.interrupt() for our id mid-probe.
                 drop(timeout_guard);
                 clear_stale_interrupt(&conn).await;
-                let _retry_guard = register_remaining_timeout(&conn, deadline)?;
+                let _retry_guard = register_timeout(&conn, self.query_timeout);
                 conn.prepare(&sql).await.map_err(Error::from)?
             }
             Err(err) => return Err(Error::from(err).into()),
@@ -425,13 +419,7 @@ impl Database {
             pluck: false.into(),
             timing: false.into(),
         };
-        Ok(Statement::new(
-            conn,
-            stmt,
-            mode,
-            self.query_timeout,
-            self.execution_lock.clone(),
-        ))
+        Ok(Statement::new(conn, stmt, mode, self.query_timeout))
     }
 
     /// Sets the authorizer for the database.
@@ -573,9 +561,7 @@ impl Database {
             Some(timeout_ms) => query_timeout_duration(timeout_ms),
             None => self.query_timeout,
         };
-        let (_execution_guard, deadline) =
-            acquire_execution_lock(&self.execution_lock, query_timeout).await?;
-        let _timeout_guard = register_remaining_timeout(&conn, deadline)?;
+        let _timeout_guard = register_timeout(&conn, query_timeout);
         conn.execute_batch(&sql).await.map_err(Error::from)?;
         Ok(())
     }
@@ -865,58 +851,18 @@ fn is_sqlite_interrupt(err: &libsql::Error) -> bool {
     )
 }
 
-fn query_timeout_error() -> napi::Error {
-    throw_sqlite_error(
-        "interrupted".to_string(),
-        "SQLITE_INTERRUPT".to_string(),
-        libsql::ffi::SQLITE_INTERRUPT,
-    )
-}
-
-fn timeout_deadline(timeout: Duration) -> Instant {
-    let now = Instant::now();
-    now.checked_add(timeout)
-        .unwrap_or_else(|| now + Duration::from_secs(86400))
-}
-
-fn register_remaining_timeout(
-    conn: &Arc<libsql::Connection>,
-    deadline: Option<Instant>,
-) -> Result<Option<QueryTimeoutGuard>> {
-    match deadline {
-        Some(deadline) => match deadline.checked_duration_since(Instant::now()) {
-            Some(remaining) if remaining > Duration::ZERO => {
-                Ok(Some(QueryTimeoutManager::global().register(conn, remaining)))
-            }
-            _ => Err(query_timeout_error()),
-        },
-        None => Ok(None),
-    }
-}
 
-/// Acquire the per-connection execution lock, enforcing the query timeout
-/// across the queue-wait. Returns the lock guard and the absolute deadline
-/// for the current operation. If the timeout elapses while waiting, returns
-/// a SQLITE_INTERRUPT error so the caller sees the same behaviour as when a
-/// timeout fires during execution.
-async fn acquire_execution_lock(
-    lock: &Arc<tokio::sync::Mutex<()>>,
+/// Register a query timeout against `target` for the operation about to run,
+/// returning a guard that cancels the timeout when dropped. The target is the
+/// statement for per-statement operations (so the timer interrupts only that
+/// statement, leaving concurrent operations on the same connection untouched)
+/// or the connection for connection-wide operations. Returns `None` when no
+/// timeout is in effect.
+fn register_timeout<T: query_timeout::Interruptible + 'static>(
+    target: &Arc<T>,
     timeout: Option<Duration>,
-) -> Result<(tokio::sync::OwnedMutexGuard<()>, Option<Instant>)> {
-    let lock = lock.clone();
-    match timeout {
-        None => Ok((lock.lock_owned().await, None)),
-        Some(t) => {
-            let deadline = timeout_deadline(t);
-            match tokio::time::timeout(t, lock.lock_owned()).await {
-                Ok(guard) => match deadline.checked_duration_since(Instant::now()) {
-                    Some(remaining) if remaining > Duration::ZERO => Ok((guard, Some(deadline))),
-                    _ => Err(query_timeout_error()),
-                },
-                Err(_) => Err(query_timeout_error()),
-            }
-        }
-    }
+) -> Option<QueryTimeoutGuard> {
+    timeout.map(|t| QueryTimeoutManager::global().register(target, t))
 }
 
 async fn clear_stale_interrupt(conn: &Arc<libsql::Connection>) {
@@ -944,8 +890,6 @@ pub struct Statement {
     mode: AccessMode,
     // Maximum time in milliseconds that a query is allowed to run.
     query_timeout: Option<Duration>,
-    // Shared per-connection execution lock.
-    execution_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 #[napi]
@@ -962,7 +906,6 @@ impl Statement {
         stmt: libsql::Statement,
         mode: AccessMode,
         query_timeout: Option<Duration>,
-        execution_lock: Arc<tokio::sync::Mutex<()>>,
     ) -> Self {
         let column_names: Vec<std::ffi::CString> = stmt
             .columns()
@@ -976,7 +919,6 @@ impl Statement {
             column_names,
             mode,
             query_timeout,
-            execution_lock,
         }
     }
 
@@ -999,12 +941,9 @@ impl Statement {
         let stmt = self.stmt.clone();
         let conn = self.conn.clone();
         let query_timeout = self.resolve_query_timeout(query_options);
-        let execution_lock = self.execution_lock.clone();
 
         let future = async move {
-            let (_execution_guard, deadline) =
-                acquire_execution_lock(&execution_lock, query_timeout).await?;
-            let _timeout_guard = register_remaining_timeout(&conn, deadline)?;
+            let _timeout_guard = register_timeout(&stmt, query_timeout);
             stmt.run(params).await.map_err(Error::from)?;
             let changes = if conn.total_changes() == total_changes_before {
                 0
@@ -1052,14 +991,10 @@ impl Statement {
 
         let stmt = self.stmt.clone();
         let stmt_fut = stmt.clone();
-        let conn = self.conn.clone();
         let query_timeout = self.resolve_query_timeout(query_options);
-        let execution_lock = self.execution_lock.clone();
         let future = async move {
-            let (_execution_guard, deadline) =
-                acquire_execution_lock(&execution_lock, query_timeout).await?;
             let result: std::result::Result<(Option<libsql::Row>, Option<f64>), Error> = {
-                let _timeout_guard = register_remaining_timeout(&conn, deadline)?;
+                let _timeout_guard = register_timeout(&stmt_fut, query_timeout);
                 async {
                     let mut rows = stmt_fut.query(params).await.map_err(Error::from)?;
                     let row = rows.next().await.map_err(Error::from)?;
@@ -1137,32 +1072,24 @@ impl Statement {
         let params = map_params(&stmt, params)?;
         let stmt_for_query = self.stmt.clone();
         let stmt_for_iter = stmt_for_query.clone();
-        let conn = self.conn.clone();
         let query_timeout = self.resolve_query_timeout(query_options);
-        let execution_lock = self.execution_lock.clone();
         let future = async move {
-            let (execution_guard, deadline) =
-                acquire_execution_lock(&execution_lock, query_timeout).await?;
-            let timeout_guard = register_remaining_timeout(&conn, deadline)?;
+            let timeout_guard = register_timeout(&stmt_for_query, query_timeout);
             let rows = stmt_for_query.query(params).await.map_err(Error::from)?;
-            Ok::<_, napi::Error>((rows, execution_guard, timeout_guard))
+            Ok::<_, napi::Error>((rows, timeout_guard))
         };
         let column_names = self.column_names.clone();
-        env.execute_tokio_future(
-            future,
-            move |&mut _env, (result, execution_guard, timeout_guard)| {
-                Ok(RowsIterator::new(
-                    Arc::new(tokio::sync::Mutex::new(result)),
-                    stmt_for_iter,
-                    column_names,
-                    safe_ints,
-                    raw,
-                    pluck,
-                    timeout_guard,
-                    execution_guard,
-                ))
-            },
-        )
+        env.execute_tokio_future(future, move |&mut _env, (result, timeout_guard)| {
+            Ok(RowsIterator::new(
+                Arc::new(tokio::sync::Mutex::new(result)),
+                stmt_for_iter,
+                column_names,
+                safe_ints,
+                raw,
+                pluck,
+                timeout_guard,
+            ))
+        })
     }
 
     #[napi]
@@ -1274,12 +1201,9 @@ pub fn statement_get_sync(
 
     let rt = runtime()?;
     let query_timeout = stmt.resolve_query_timeout(query_options);
-    let execution_lock = stmt.execution_lock.clone();
     let result: Result<(Option<libsql::Row>, Option<f64>)> = {
         rt.block_on(async move {
-            let (_execution_guard, deadline) =
-                acquire_execution_lock(&execution_lock, query_timeout).await?;
-            let _timeout_guard = register_remaining_timeout(&stmt.conn, deadline)?;
+            let _timeout_guard = register_timeout(&stmt.stmt, query_timeout);
             let params = map_params(&stmt.stmt, params)?;
             let mut rows = stmt.stmt.query(params).await.map_err(Error::from)?;
             let row = rows.next().await.map_err(Error::from)?;
@@ -1318,11 +1242,8 @@ pub fn statement_run_sync(
     stmt.stmt.reset();
     let rt = runtime()?;
     let query_timeout = stmt.resolve_query_timeout(query_options);
-    let execution_lock = stmt.execution_lock.clone();
     rt.block_on(async move {
-        let (_execution_guard, deadline) =
-            acquire_execution_lock(&execution_lock, query_timeout).await?;
-        let _timeout_guard = register_remaining_timeout(&stmt.conn, deadline)?;
+        let _timeout_guard = register_timeout(&stmt.stmt, query_timeout);
         let params = map_params(&stmt.stmt, params)?;
         let total_changes_before = stmt.conn.total_changes();
         let start = std::time::Instant::now();
@@ -1355,14 +1276,10 @@ pub fn statement_iterate_sync(
     let raw = stmt.mode.raw.load(Ordering::SeqCst);
     let pluck = stmt.mode.pluck.load(Ordering::SeqCst);
     let query_timeout = stmt.resolve_query_timeout(query_options);
-    let conn = stmt.conn.clone();
-    let execution_lock = stmt.execution_lock.clone();
     let inner_stmt = stmt.stmt.clone();
     let iter_stmt = inner_stmt.clone();
-    let (rows, column_names, execution_guard, timeout_guard) = rt.block_on(async move {
-        let (execution_guard, deadline) =
-            acquire_execution_lock(&execution_lock, query_timeout).await?;
-        let timeout_guard = register_remaining_timeout(&conn, deadline)?;
+    let (rows, column_names, timeout_guard) = rt.block_on(async move {
+        let timeout_guard = register_timeout(&inner_stmt, query_timeout);
         inner_stmt.reset();
         let params = map_params(&inner_stmt, params)?;
         let rows = inner_stmt.query(params).await.map_err(Error::from)?;
@@ -1371,7 +1288,7 @@ pub fn statement_iterate_sync(
             column_names
                 .push(std::ffi::CString::new(rows.column_name(i).unwrap().to_string()).unwrap());
         }
-        Ok::<_, napi::Error>((rows, column_names, execution_guard, timeout_guard))
+        Ok::<_, napi::Error>((rows, column_names, timeout_guard))
     })?;
     Ok(RowsIterator::new(
         Arc::new(tokio::sync::Mutex::new(rows)),
@@ -1381,7 +1298,6 @@ pub fn statement_iterate_sync(
         raw,
         pluck,
         timeout_guard,
-        execution_guard,
     ))
 }
 
@@ -1530,7 +1446,6 @@ pub struct RowsIterator {
     raw: bool,
     pluck: bool,
     timeout_guard: Mutex<Option<QueryTimeoutGuard>>,
-    execution_guard: Mutex<Option<tokio::sync::OwnedMutexGuard<()>>>,
 }
 
 #[napi]
@@ -1543,7 +1458,6 @@ impl RowsIterator {
         raw: bool,
         pluck: bool,
         timeout_guard: Option<QueryTimeoutGuard>,
-        execution_guard: tokio::sync::OwnedMutexGuard<()>,
     ) -> Self {
         Self {
             rows,
@@ -1553,7 +1467,6 @@ impl RowsIterator {
             raw,
             pluck,
             timeout_guard: Mutex::new(timeout_guard),
-            execution_guard: Mutex::new(Some(execution_guard)),
         }
     }
 
@@ -1588,8 +1501,6 @@ impl RowsIterator {
         self.stmt.reset();
         let mut timeout_guard = self.timeout_guard.lock().unwrap();
         timeout_guard.take();
-        let mut execution_guard = self.execution_guard.lock().unwrap();
-        execution_guard.take();
     }
 }