feat: add configurable timeout for interruptible transactions

begin_interruptible_transaction holds the exclusive writer until the
frontend commits or rolls back. If the frontend never does, the writer
is locked indefinitely. Add a configurable idle timeout (default 5 min)
with lazy cleanup: expired transactions are evicted on the next insert
or rejected on the next remove, auto-rolling back via Drop.

Convert ActiveInterruptibleTransactions from a tuple struct to named
fields with a timeout Duration, and add a created_at Instant to each
ActiveInterruptibleTransaction. Expose Builder::transaction_timeout()
in the plugin for configuration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pmorris-dev

crates/sqlx-sqlite-toolkit/src/error.rs

@@ -45,6 +45,10 @@ pub enum Error {
    #[error("invalid transaction token")]
    InvalidTransactionToken,
 
+   /// Transaction timed out (exceeded the configured idle timeout).
+   #[error("transaction timed out for database: {0}")]
+   TransactionTimedOut(String),
+
    /// Error from the observer (change notifications).
    #[cfg(feature = "observer")]
    #[error(transparent)]
@@ -115,6 +119,7 @@ impl Error {
          Error::TransactionAlreadyActive(_) => "TRANSACTION_ALREADY_ACTIVE".to_string(),
          Error::NoActiveTransaction(_) => "NO_ACTIVE_TRANSACTION".to_string(),
          Error::InvalidTransactionToken => "INVALID_TRANSACTION_TOKEN".to_string(),
+         Error::TransactionTimedOut(_) => "TRANSACTION_TIMED_OUT".to_string(),
          #[cfg(feature = "observer")]
          Error::Observer(_) => "OBSERVER_ERROR".to_string(),
          Error::Io(_) => "IO_ERROR".to_string(),
@@ -194,6 +199,13 @@ mod tests {
       assert_eq!(err.error_code(), "IO_ERROR");
    }
 
+   #[test]
+   fn test_error_code_transaction_timed_out() {
+      let err = Error::TransactionTimedOut("test.db".into());
+      assert_eq!(err.error_code(), "TRANSACTION_TIMED_OUT");
+      assert!(err.to_string().contains("test.db"));
+   }
+
    #[test]
    fn test_error_code_other() {
       let err = Error::Other("something went wrong".into());

crates/sqlx-sqlite-toolkit/src/transactions.rs

@@ -2,6 +2,7 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::{Duration, Instant};
 
 use indexmap::IndexMap;
 use serde::Deserialize;
@@ -10,7 +11,7 @@ use sqlx::{Column, Row};
 use sqlx_sqlite_conn_mgr::{AttachedWriteGuard, WriteGuard};
 use tokio::sync::{Mutex, RwLock};
 use tokio::task::AbortHandle;
-use tracing::debug;
+use tracing::{debug, warn};
 
 #[cfg(feature = "observer")]
 use sqlx_sqlite_observer::ObservableWriteGuard;
@@ -97,6 +98,7 @@ pub struct ActiveInterruptibleTransaction {
    db_path: String,
    transaction_id: String,
    writer: Option<TransactionWriter>,
+   created_at: Instant,
 }
 
 impl ActiveInterruptibleTransaction {
@@ -105,6 +107,7 @@ impl ActiveInterruptibleTransaction {
          db_path,
          transaction_id,
          writer: Some(writer),
+         created_at: Instant::now(),
       }
    }
 
@@ -241,34 +244,71 @@ impl Drop for ActiveInterruptibleTransaction {
    }
 }
 
+/// Default transaction timeout (5 minutes).
+const DEFAULT_TRANSACTION_TIMEOUT: Duration = Duration::from_secs(300);
+
 /// Global state tracking all active interruptible transactions.
 ///
-/// Enforces one interruptible transaction per database path.
+/// Enforces one interruptible transaction per database path and applies a configurable
+/// idle timeout. Expired transactions are cleaned up lazily on the next `insert()` or
+/// `remove()` call — no background task is needed.
+///
 /// Uses `Mutex` rather than `RwLock` because all operations require write access,
 /// and `Mutex<T>` only requires `T: Send` (not `T: Sync`) — avoiding an
 /// `unsafe impl Sync` that would otherwise be needed due to non-`Sync` inner
 /// types (`PoolConnection`, raw pointers in observer guards).
-#[derive(Clone, Default)]
-pub struct ActiveInterruptibleTransactions(
-   Arc<Mutex<HashMap<String, ActiveInterruptibleTransaction>>>,
-);
+#[derive(Clone)]
+pub struct ActiveInterruptibleTransactions {
+   inner: Arc<Mutex<HashMap<String, ActiveInterruptibleTransaction>>>,
+   timeout: Duration,
+}
+
+impl Default for ActiveInterruptibleTransactions {
+   fn default() -> Self {
+      Self::new(DEFAULT_TRANSACTION_TIMEOUT)
+   }
+}
 
 impl ActiveInterruptibleTransactions {
+   /// Create a new instance with the given transaction timeout.
+   pub fn new(timeout: Duration) -> Self {
+      Self {
+         inner: Arc::new(Mutex::new(HashMap::new())),
+         timeout,
+      }
+   }
+
    pub async fn insert(&self, db_path: String, tx: ActiveInterruptibleTransaction) -> Result<()> {
       use std::collections::hash_map::Entry;
-      let mut txs = self.0.lock().await;
+      let mut txs = self.inner.lock().await;
 
       match txs.entry(db_path.clone()) {
          Entry::Vacant(e) => {
             e.insert(tx);
             Ok(())
          }
-         Entry::Occupied(_) => Err(Error::TransactionAlreadyActive(db_path)),
+         Entry::Occupied(mut e) => {
+            // If the existing transaction has expired, drop it (auto-rollback) and
+            // replace with the new one.
+            if e.get().created_at.elapsed() >= self.timeout {
+               warn!(
+                  "Evicting expired transaction for db: {} (age: {:?}, timeout: {:?})",
+                  db_path,
+                  e.get().created_at.elapsed(),
+                  self.timeout,
+               );
+               // Drop the expired transaction (auto-rollback) before inserting the new one
+               let _expired = e.insert(tx);
+               Ok(())
+            } else {
+               Err(Error::TransactionAlreadyActive(db_path))
+            }
+         }
       }
    }
 
    pub async fn abort_all(&self) {
-      let mut txs = self.0.lock().await;
+      let mut txs = self.inner.lock().await;
       debug!("Aborting {} active interruptible transaction(s)", txs.len());
 
       for db_path in txs.keys() {
@@ -283,13 +323,17 @@ impl ActiveInterruptibleTransactions {
       txs.clear();
    }
 
-   /// Remove and return transaction for commit/rollback
+   /// Remove and return transaction for commit/rollback.
+   ///
+   /// Returns `Err(Error::TransactionTimedOut)` if the transaction has exceeded the
+   /// configured timeout. The expired transaction is dropped (auto-rolled-back) in
+   /// that case.
    pub async fn remove(
       &self,
       db_path: &str,
       token_id: &str,
    ) -> Result<ActiveInterruptibleTransaction> {
-      let mut txs = self.0.lock().await;
+      let mut txs = self.inner.lock().await;
 
       // Validate token before removal
       let tx = txs
@@ -300,6 +344,19 @@ impl ActiveInterruptibleTransactions {
          return Err(Error::InvalidTransactionToken);
       }
 
+      // Check if the transaction has expired
+      if tx.created_at.elapsed() >= self.timeout {
+         warn!(
+            "Transaction timed out for db: {} (age: {:?}, timeout: {:?})",
+            db_path,
+            tx.created_at.elapsed(),
+            self.timeout,
+         );
+         // Drop the expired transaction (auto-rollback via Drop)
+         txs.remove(db_path);
+         return Err(Error::TransactionTimedOut(db_path.to_string()));
+      }
+
       // Safe unwrap: we just confirmed the key exists above
       Ok(txs.remove(db_path).unwrap())
    }

crates/sqlx-sqlite-toolkit/tests/transaction_state_tests.rs

@@ -202,6 +202,80 @@ async fn test_insert_after_abort_all_succeeds() {
    state.insert("reuse-key".into(), tx2).await.unwrap();
 }
 
+// ============================================================================
+// ActiveInterruptibleTransactions timeout tests
+// ============================================================================
+
+#[tokio::test]
+async fn test_expired_transaction_evicted_on_insert() {
+   let (db1, _temp1) = create_test_db("expire1.db").await;
+   let (db2, _temp2) = create_test_db("expire2.db").await;
+
+   for db in [&db1, &db2] {
+      db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)".into(), vec![])
+         .await
+         .unwrap();
+   }
+
+   // Use a 1ms timeout so the first transaction expires immediately
+   let state = ActiveInterruptibleTransactions::new(std::time::Duration::from_millis(1));
+
+   let tx1 = begin_transaction(&db1, "shared-key").await;
+   state.insert("shared-key".into(), tx1).await.unwrap();
+
+   // Sleep to ensure the transaction expires
+   tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+   // Second insert should succeed because the expired transaction is evicted
+   let tx2 = begin_transaction(&db2, "shared-key").await;
+   state.insert("shared-key".into(), tx2).await.unwrap();
+}
+
+#[tokio::test]
+async fn test_remove_expired_transaction_returns_timed_out() {
+   let (db, _temp) = create_test_db("timeout.db").await;
+
+   db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)".into(), vec![])
+      .await
+      .unwrap();
+
+   let state = ActiveInterruptibleTransactions::new(std::time::Duration::from_millis(1));
+
+   let tx = begin_transaction(&db, "timeout.db").await;
+   let tx_id = tx.transaction_id().to_string();
+
+   state.insert("timeout.db".into(), tx).await.unwrap();
+
+   // Sleep to ensure the transaction expires
+   tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+   let err = expect_err(state.remove("timeout.db", &tx_id).await);
+   assert_eq!(err.error_code(), "TRANSACTION_TIMED_OUT");
+}
+
+#[tokio::test]
+async fn test_non_expired_transaction_not_evicted() {
+   let (db1, _temp1) = create_test_db("live1.db").await;
+   let (db2, _temp2) = create_test_db("live2.db").await;
+
+   for db in [&db1, &db2] {
+      db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)".into(), vec![])
+         .await
+         .unwrap();
+   }
+
+   // Use a long timeout so the first transaction does NOT expire
+   let state = ActiveInterruptibleTransactions::new(std::time::Duration::from_secs(300));
+
+   let tx1 = begin_transaction(&db1, "shared-key").await;
+   state.insert("shared-key".into(), tx1).await.unwrap();
+
+   // Second insert should still fail because the first transaction is alive
+   let tx2 = begin_transaction(&db2, "shared-key").await;
+   let err = state.insert("shared-key".into(), tx2).await.unwrap_err();
+   assert_eq!(err.error_code(), "TRANSACTION_ALREADY_ACTIVE");
+}
+
 // ============================================================================
 // ActiveRegularTransactions tests
 // ============================================================================

src/lib.rs

@@ -134,13 +134,16 @@ pub struct MigrationEvent {
 pub struct Builder {
    /// Migrations registered per database path
    migrations: HashMap<String, Arc<Migrator>>,
+   /// Timeout for interruptible transactions. Defaults to 5 minutes.
+   transaction_timeout: Option<std::time::Duration>,
 }
 
 impl Builder {
    /// Create a new builder instance.
    pub fn new() -> Self {
       Self {
          migrations: HashMap::new(),
+         transaction_timeout: None,
       }
    }
 
@@ -170,9 +173,19 @@ impl Builder {
       self
    }
 
+   /// Set the timeout for interruptible transactions.
+   ///
+   /// If an interruptible transaction exceeds this duration, it will be automatically
+   /// rolled back on the next access attempt. Defaults to 5 minutes.
+   pub fn transaction_timeout(mut self, timeout: std::time::Duration) -> Self {
+      self.transaction_timeout = Some(timeout);
+      self
+   }
+
    /// Build the plugin with command registration and state management.
    pub fn build<R: Runtime>(self) -> tauri::plugin::TauriPlugin<R> {
       let migrations = Arc::new(self.migrations);
+      let transaction_timeout = self.transaction_timeout;
 
       PluginBuilder::<R>::new("sqlite")
          .invoke_handler(tauri::generate_handler![
@@ -197,7 +210,10 @@ impl Builder {
          .setup(move |app, _api| {
             app.manage(DbInstances::default());
             app.manage(MigrationStates::default());
-            app.manage(ActiveInterruptibleTransactions::default());
+            app.manage(match transaction_timeout {
+               Some(timeout) => ActiveInterruptibleTransactions::new(timeout),
+               None => ActiveInterruptibleTransactions::default(),
+            });
             app.manage(ActiveRegularTransactions::default());
             app.manage(subscriptions::ActiveSubscriptions::default());