Merge 'bindings/python: close connection only when reference count is one' from Pere Diaz Bou

Due to how `execute` is implemented, it returns a `Connection` clone
which internally shares a turso_core::Connection with every other
Connection. Since `execute` returns `Connection` and immediatly it is
dropped, it will close connection, checkpoint and leave database in
weird state.
Let's make sure we don't keep using a connection after it was dropped.
In case of executing a query that was closed we will try to rollback and
return early.

Closes #2006

bindings/python/src/lib.rs

@@ -296,9 +296,11 @@ impl Connection {
 
 impl Drop for Connection {
     fn drop(&mut self) {
-        self.conn
-            .close()
-            .expect("Failed to drop (close) connection");
+        if Arc::strong_count(&self.conn) == 1 {
+            self.conn
+                .close()
+                .expect("Failed to drop (close) connection");
+        }
     }
 }
 

core/lib.rs

@@ -280,6 +280,7 @@ impl Database {
                 readonly: Cell::new(false),
                 wal_checkpoint_disabled: Cell::new(false),
                 capture_data_changes: RefCell::new(CaptureDataChangesMode::Off),
+                closed: Cell::new(false),
             });
             if let Err(e) = conn.register_builtins() {
                 return Err(LimboError::ExtensionError(e));
@@ -333,6 +334,7 @@ impl Database {
             readonly: Cell::new(false),
             wal_checkpoint_disabled: Cell::new(false),
             capture_data_changes: RefCell::new(CaptureDataChangesMode::Off),
+            closed: Cell::new(false),
         });
 
         if let Err(e) = conn.register_builtins() {
@@ -487,6 +489,7 @@ pub struct Connection {
     readonly: Cell<bool>,
     wal_checkpoint_disabled: Cell<bool>,
     capture_data_changes: RefCell<CaptureDataChangesMode>,
+    closed: Cell<bool>,
 }
 
 impl Connection {
@@ -739,6 +742,8 @@ impl Connection {
 
     /// Close a connection and checkpoint.
     pub fn close(&self) -> Result<()> {
+        turso_assert!(!self.closed.get(), "Connection already closed");
+        self.closed.set(true);
         self.pager
             .checkpoint_shutdown(self.wal_checkpoint_disabled.get())
     }

core/vdbe/mod.rs

@@ -380,6 +380,14 @@ impl Program {
         pager: Rc<Pager>,
     ) -> Result<StepResult> {
         loop {
+            if *self.connection.closed.borrow() {
+                // Connection is closed for whatever reason, rollback the transaction.
+                let state = self.connection.transaction_state.get();
+                if let TransactionState::Write { schema_did_change } = state {
+                    pager.rollback(schema_did_change, &self.connection)?
+                }
+                return Err(LimboError::InternalError("Connection closed".to_string()));
+            }
             if state.is_interrupted() {
                 return Ok(StepResult::Interrupt);
             }