fix: rollback on drop so write conn returns to pool clean (#46)

Dropping an ActiveInterruptibleTransaction without commit/rollback left
the write connection in the pool with an open transaction. sqlx pools
reuse connections rather than closing them, so SQLite's close-time
auto-rollback never fired, and the next acquire_writer() got a
connection where BEGIN IMMEDIATE failed with "cannot start a
transaction within a transaction".

Drop now takes the writer and spawns a tokio task that issues ROLLBACK
(and detach_if_attached) before the connection is released to the pool.
The insert()/remove()/abort_all() paths that previously relied on Drop
now roll back explicitly.

As defense-in-depth against any other writer path that might leak a
transaction (e.g. user code that does BEGIN and returns early), the
write pool gets an after_release hook that unconditionally runs
ROLLBACK, ignoring the benign "no transaction is active" error.

Adds a regression test that mirrors the reporter's MCVE: start an
interruptible transaction, drop it, start a second one, and assert the
second succeeds and only its data commits.

Updates the README to document the auto-rollback-on-drop guarantee
(including early returns via ?).

Author: pmorris-dev

README.md

@@ -381,8 +381,10 @@ await tx.commit();
    * Only one interruptible transaction can be active per database at a time
    * The write lock is held for the entire duration - keep transactions short
    * Uncommitted writes are visible only within the transaction's `read()` method
-   * Always commit or rollback - abandoned transactions will rollback automatically
-     on app exit
+   * If the transaction handle is dropped without calling `commit()` or
+     `rollback()`, the transaction is automatically rolled back and the write
+     connection is released back to the pool. This also happens on app exit
+     and on transaction timeout.
 
 To rollback instead of committing:
 
@@ -784,6 +786,10 @@ println!("Transaction completed: {} statements executed", results.len());
 
 For transactions that need to read data mid-transaction:
 
+If `tx` is dropped without calling `commit()` or `rollback()` — including via
+an early return from a `?` operator — the transaction is automatically rolled
+back and the write connection is released back to the pool.
+
 ```rust
 // Assuming user_id, product_id, item_total are defined in your application context
 let user_id = 123;

crates/sqlx-sqlite-conn-mgr/src/database.rs

@@ -177,12 +177,26 @@ impl SqliteDatabase {
             .read_only(false)
             .optimize_on_close(true, OPTIMIZE_ANALYSIS_LIMIT);
 
+         // Defense-in-depth: when any writer is returned to the pool, issue
+         // ROLLBACK to discard any transaction that a caller may have left open
+         // (e.g., a writer dropped after BEGIN without COMMIT/ROLLBACK). SQLite
+         // only auto-rollbacks on connection close, not on pool return, so
+         // without this the next acquire_writer() sees "cannot start a
+         // transaction within a transaction". The "cannot rollback - no
+         // transaction is active" error on a clean connection is benign and
+         // ignored.
          let write_conn = SqlitePoolOptions::new()
             .max_connections(1)
             .min_connections(0)
             .idle_timeout(Some(std::time::Duration::from_secs(
                config.idle_timeout_secs,
             )))
+            .after_release(|conn, _meta| {
+               Box::pin(async move {
+                  let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
+                  Ok(true)
+               })
+            })
             .connect_with(write_options)
             .await?;
 

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

@@ -232,14 +232,47 @@ impl From<(String, Vec<JsonValue>)> for Statement {
 
 impl Drop for ActiveInterruptibleTransaction {
    fn drop(&mut self) {
-      // If writer is still present, it means commit/rollback wasn't called.
-      // SQLite will automatically ROLLBACK the transaction when the connection
-      // is returned to the pool if no explicit COMMIT was issued.
-      if self.writer.is_some() {
-         debug!(
-            "Dropping transaction for db: {}, tx_id: {} (will auto-rollback)",
-            self.db_path, self.transaction_id
-         );
+      // If writer is still present, commit/rollback was not called. The connection
+      // is about to return to the pool — we must issue ROLLBACK explicitly because
+      // sqlx pools reuse the connection (SQLite only auto-rollbacks on close, not
+      // on pool return). Without this, the next acquire_writer() gets a connection
+      // with an open transaction and "BEGIN IMMEDIATE" fails.
+      let Some(mut writer) = self.writer.take() else {
+         return;
+      };
+      let db_path = std::mem::take(&mut self.db_path);
+      let tx_id = std::mem::take(&mut self.transaction_id);
+
+      match tokio::runtime::Handle::try_current() {
+         Ok(handle) => {
+            debug!(
+               "Dropping transaction for db: {}, tx_id: {} (auto-rollback scheduled)",
+               db_path, tx_id
+            );
+            handle.spawn(async move {
+               if let Err(e) = writer.rollback().await {
+                  warn!(
+                     "auto-rollback on drop failed (db: {}, tx: {}): {}",
+                     db_path, tx_id, e
+                  );
+               }
+               if let Err(e) = writer.detach_if_attached().await {
+                  warn!(
+                     "detach_all after auto-rollback failed (db: {}, tx: {}): {}",
+                     db_path, tx_id, e
+                  );
+               }
+               // writer drops here — connection returns to pool clean
+            });
+         }
+         Err(_) => {
+            warn!(
+               "no tokio runtime at drop for db: {}, tx: {} — write pool after_release hook will clean up",
+               db_path, tx_id
+            );
+            // Writer drops synchronously; the write pool's after_release hook will
+            // issue ROLLBACK before the connection is handed out again.
+         }
       }
    }
 }
@@ -288,17 +321,21 @@ impl ActiveInterruptibleTransactions {
             Ok(())
          }
          Entry::Occupied(mut e) => {
-            // If the existing transaction has expired, drop it (auto-rollback) and
-            // replace with the new one.
+            // If the existing transaction has expired, roll it back and replace
+            // with the new one. We rollback explicitly (rather than relying on
+            // Drop) so the writer is guaranteed to return to the pool clean
+            // before the caller tries to start a new transaction on it.
             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);
+               let expired = e.insert(tx);
+               if let Err(err) = expired.rollback().await {
+                  warn!("rollback of expired transaction failed (db: {db_path}): {err}");
+               }
                Ok(())
             } else {
                Err(Error::TransactionAlreadyActive(db_path))
@@ -308,57 +345,68 @@ impl ActiveInterruptibleTransactions {
    }
 
    pub async fn abort_all(&self) {
-      let mut txs = self.inner.lock().await;
-      debug!("Aborting {} active interruptible transaction(s)", txs.len());
-
-      for db_path in txs.keys() {
+      // Drain under the lock, then release it before awaiting rollbacks so we
+      // don't hold the mutex across network of awaits.
+      let drained: Vec<(String, ActiveInterruptibleTransaction)> = {
+         let mut txs = self.inner.lock().await;
+         debug!("Aborting {} active interruptible transaction(s)", txs.len());
+         txs.drain().collect()
+      };
+
+      for (db_path, tx) in drained {
          debug!(
-            "Dropping interruptible transaction for database: {}",
+            "Rolling back interruptible transaction for database: {}",
             db_path
          );
+         if let Err(err) = tx.rollback().await {
+            warn!("rollback during abort_all failed (db: {db_path}): {err}");
+         }
       }
-
-      // Clear all transactions to drop WriteGuards and release locks
-      // Dropping triggers auto-rollback via Drop trait
-      txs.clear();
    }
 
    /// 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.
+   /// configured timeout. The expired transaction is rolled back before the error
+   /// is returned.
    pub async fn remove(
       &self,
       db_path: &str,
       token_id: &str,
    ) -> Result<ActiveInterruptibleTransaction> {
-      let mut txs = self.inner.lock().await;
+      // Take the expired transaction out under the lock, then rollback without
+      // holding it so other callers don't block on an unrelated cleanup.
+      let expired: Option<ActiveInterruptibleTransaction> = {
+         let mut txs = self.inner.lock().await;
 
-      // Validate token before removal
-      let tx = txs
-         .get(db_path)
-         .ok_or_else(|| Error::NoActiveTransaction(db_path.to_string()))?;
+         let tx = txs
+            .get(db_path)
+            .ok_or_else(|| Error::NoActiveTransaction(db_path.to_string()))?;
 
-      if tx.transaction_id() != token_id {
-         return Err(Error::InvalidTransactionToken);
-      }
+         if tx.transaction_id() != token_id {
+            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()));
-      }
+         if tx.created_at.elapsed() >= self.timeout {
+            warn!(
+               "Transaction timed out for db: {} (age: {:?}, timeout: {:?})",
+               db_path,
+               tx.created_at.elapsed(),
+               self.timeout,
+            );
+            txs.remove(db_path)
+         } else {
+            // Safe unwrap: we just confirmed the key exists above
+            return Ok(txs.remove(db_path).unwrap());
+         }
+      };
 
-      // Safe unwrap: we just confirmed the key exists above
-      Ok(txs.remove(db_path).unwrap())
+      if let Some(tx) = expired
+         && let Err(err) = tx.rollback().await
+      {
+         warn!("rollback of timed-out transaction failed (db: {db_path}): {err}");
+      }
+      Err(Error::TransactionTimedOut(db_path.to_string()))
    }
 }
 

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

@@ -377,3 +377,58 @@ async fn test_execute_transaction_rollback_on_failure() {
 
    db.remove().await.unwrap();
 }
+
+/// Regression test for issue #46:
+/// https://github.com/silvermine/tauri-plugin-sqlite/issues/46
+///
+/// Dropping an interruptible transaction without calling commit/rollback must
+/// leave the shared write connection in a clean state so the next transaction
+/// can start. Previously the writer was returned to the pool with an open
+/// transaction, causing "cannot start a transaction within a transaction" on
+/// the next BEGIN IMMEDIATE.
+#[tokio::test]
+async fn test_dropped_transaction_releases_write_connection() {
+   let (db, _temp) = create_test_db("test.db").await;
+
+   db.execute(
+      "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)".into(),
+      vec![],
+   )
+   .await
+   .unwrap();
+
+   // Start a transaction and drop it without commit/rollback.
+   {
+      let _tx = db
+         .begin_interruptible_transaction()
+         .execute(vec![(
+            "INSERT INTO users (name) VALUES (?)",
+            vec![json!("Alice")],
+         )])
+         .await
+         .unwrap();
+   }
+
+   // Starting a second interruptible transaction must succeed — the previous
+   // writer should have been rolled back and returned to the pool clean.
+   let tx2 = db
+      .begin_interruptible_transaction()
+      .execute(vec![(
+         "INSERT INTO users (name) VALUES (?)",
+         vec![json!("Bob")],
+      )])
+      .await
+      .expect("second transaction should start on a clean connection");
+
+   tx2.commit().await.unwrap();
+
+   // Only Bob should be present — Alice's insert was rolled back on drop.
+   let rows = db
+      .fetch_all("SELECT name FROM users ORDER BY id".into(), vec![])
+      .await
+      .unwrap();
+   assert_eq!(rows.len(), 1);
+   assert_eq!(rows[0].get("name").and_then(|v| v.as_str()), Some("Bob"));
+
+   db.remove().await.unwrap();
+}