Merge pull request #367 from kornelski/master

Roll back uncommitable transactions

src/lib.rs

@@ -623,7 +623,7 @@ bitflags! {
 
 impl Default for OpenFlags {
     fn default() -> OpenFlags {
-        OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE | 
+        OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE |
         OpenFlags::SQLITE_OPEN_NO_MUTEX | OpenFlags::SQLITE_OPEN_URI
     }
 }
@@ -990,6 +990,41 @@ mod test {
         Connection::open_in_memory().unwrap()
     }
 
+    #[test]
+    fn test_concurrent_transactions_busy_commit() {
+        let tmp = TempDir::new("locked").unwrap();
+        let path = tmp.path().join("transactions.db3");
+
+        Connection::open(&path).expect("create temp db").execute_batch("
+            BEGIN; CREATE TABLE foo(x INTEGER);
+            INSERT INTO foo VALUES(42); END;")
+        .expect("create temp db");
+
+        let mut db1 = Connection::open(&path).unwrap();
+        let mut db2 = Connection::open(&path).unwrap();
+
+        db1.execute_batch("PRAGMA busy_timeout = 0;").unwrap();
+        db2.execute_batch("PRAGMA busy_timeout = 0;").unwrap();
+
+        {
+            let tx1 = db1.transaction().unwrap();
+            let tx2 = db2.transaction().unwrap();
+
+            // SELECT first makes sqlite lock with a shared lock
+            let _ = tx1.query_row("SELECT x FROM foo LIMIT 1", &[], |_| ()).unwrap();
+            let _ = tx2.query_row("SELECT x FROM foo LIMIT 1", &[], |_| ()).unwrap();
+
+            tx1.execute("INSERT INTO foo VALUES(?1)", &[&1]).unwrap();
+            let _ = tx2.execute("INSERT INTO foo VALUES(?1)", &[&2]);
+
+            let _ = tx1.commit();
+            let _ = tx2.commit();
+        }
+
+        let _ = db1.transaction().expect("commit should have closed transaction");
+        let _ = db2.transaction().expect("commit should have closed transaction");
+    }
+
     #[test]
     #[cfg_attr(rustfmt, rustfmt_skip)]
     fn test_persistence() {

src/transaction.rs

@@ -167,8 +167,9 @@ impl<'conn> Transaction<'conn> {
     }
 
     fn commit_(&mut self) -> Result<()> {
+        self.conn.execute_batch("COMMIT")?;
         self.committed = true;
-        self.conn.execute_batch("COMMIT")
+        Ok(())
     }
 
     /// A convenience method which consumes and rolls back a transaction.
@@ -177,8 +178,9 @@ impl<'conn> Transaction<'conn> {
     }
 
     fn rollback_(&mut self) -> Result<()> {
+        self.conn.execute_batch("ROLLBACK")?;
         self.committed = true;
-        self.conn.execute_batch("ROLLBACK")
+        Ok(())
     }
 
     /// Consumes the transaction, committing or rolling back according to the current setting
@@ -195,7 +197,7 @@ impl<'conn> Transaction<'conn> {
             return Ok(());
         }
         match self.drop_behavior() {
-            DropBehavior::Commit => self.commit_(),
+            DropBehavior::Commit => self.commit_().or_else(|_| self.rollback_()),
             DropBehavior::Rollback => self.rollback_(),
             DropBehavior::Ignore => Ok(()),
             DropBehavior::Panic => panic!("Transaction dropped unexpectedly."),
@@ -277,9 +279,10 @@ impl<'conn> Savepoint<'conn> {
     }
 
     fn commit_(&mut self) -> Result<()> {
-        self.committed = true;
         self.conn
-            .execute_batch(&format!("RELEASE {}", self.name))
+            .execute_batch(&format!("RELEASE {}", self.name))?;
+        self.committed = true;
+        Ok(())
     }
 
     /// A convenience method which rolls back a savepoint.
@@ -307,7 +310,7 @@ impl<'conn> Savepoint<'conn> {
             return Ok(());
         }
         match self.drop_behavior() {
-            DropBehavior::Commit => self.commit_(),
+            DropBehavior::Commit => self.commit_().or_else(|_| self.rollback()),
             DropBehavior::Rollback => self.rollback(),
             DropBehavior::Ignore => Ok(()),
             DropBehavior::Panic => panic!("Savepoint dropped unexpectedly."),