Feat: Implementation for SQLite

I have included `AtomicUsize` in `WorkerSharedState`. Ideally, it is not desirable to execute `load` and `fetch_add` in two separate steps, but we decided to allow it here since there is only one thread writing. To prevent writing from other threads, the field itself was made private, and a getter method was provided with `pub(crate)`.

sqlx-sqlite/src/connection/establish.rs

@@ -292,7 +292,6 @@ impl EstablishParams {
         Ok(ConnectionState {
             handle,
             statements: Statements::new(self.statement_cache_capacity),
-            transaction_depth: 0,
             log_settings: self.log_settings.clone(),
             progress_handler_callback: None,
             update_hook_callback: None,

sqlx-sqlite/src/connection/mod.rs

@@ -94,9 +94,6 @@ unsafe impl Send for UpdateHookHandler {}
 pub(crate) struct ConnectionState {
     pub(crate) handle: ConnectionHandle,
 
-    // transaction status
-    pub(crate) transaction_depth: usize,
-
     pub(crate) statements: Statements,
 
     log_settings: LogSettings,
@@ -211,7 +208,7 @@ impl Connection for SqliteConnection {
     }
 
     fn get_transaction_depth(&self) -> usize {
-        todo!()
+        self.worker.shared.get_transaction_depth()
     }
 
     fn cached_statements_size(&self) -> usize {

sqlx-sqlite/src/connection/worker.rs

@@ -34,10 +34,17 @@ pub(crate) struct ConnectionWorker {
 }
 
 pub(crate) struct WorkerSharedState {
+    transaction_depth: AtomicUsize,
     pub(crate) cached_statements_size: AtomicUsize,
     pub(crate) conn: Mutex<ConnectionState>,
 }
 
+impl WorkerSharedState {
+    pub(crate) fn get_transaction_depth(&self) -> usize {
+        self.transaction_depth.load(Ordering::Acquire)
+    }
+}
+
 enum Command {
     Prepare {
         query: Box<str>,
@@ -93,6 +100,7 @@ impl ConnectionWorker {
                 };
 
                 let shared = Arc::new(WorkerSharedState {
+                    transaction_depth: AtomicUsize::new(0),
                     cached_statements_size: AtomicUsize::new(0),
                     // note: must be fair because in `Command::UnlockDb` we unlock the mutex
                     // and then immediately try to relock it; an unfair mutex would immediately
@@ -181,12 +189,12 @@ impl ConnectionWorker {
                             update_cached_statements_size(&conn, &shared.cached_statements_size);
                         }
                         Command::Begin { tx } => {
-                            let depth = conn.transaction_depth;
+                            let depth = shared.transaction_depth.load(Ordering::Acquire);
                             let res =
                                 conn.handle
                                     .exec(begin_ansi_transaction_sql(depth))
                                     .map(|_| {
-                                        conn.transaction_depth += 1;
+                                        shared.transaction_depth.fetch_add(1, Ordering::Release);
                                     });
                             let res_ok = res.is_ok();
 
@@ -199,7 +207,7 @@ impl ConnectionWorker {
                                     .handle
                                     .exec(rollback_ansi_transaction_sql(depth + 1))
                                     .map(|_| {
-                                        conn.transaction_depth -= 1;
+                                        shared.transaction_depth.fetch_sub(1, Ordering::Release);
                                     })
                                 {
                                     // The rollback failed. To prevent leaving the connection
@@ -211,13 +219,13 @@ impl ConnectionWorker {
                             }
                         }
                         Command::Commit { tx } => {
-                            let depth = conn.transaction_depth;
+                            let depth = shared.transaction_depth.load(Ordering::Acquire);
 
                             let res = if depth > 0 {
                                 conn.handle
                                     .exec(commit_ansi_transaction_sql(depth))
                                     .map(|_| {
-                                        conn.transaction_depth -= 1;
+                                        shared.transaction_depth.fetch_sub(1, Ordering::Release);
                                     })
                             } else {
                                 Ok(())
@@ -237,13 +245,13 @@ impl ConnectionWorker {
                                 continue;
                             }
 
-                            let depth = conn.transaction_depth;
+                            let depth = shared.transaction_depth.load(Ordering::Acquire);
 
                             let res = if depth > 0 {
                                 conn.handle
                                     .exec(rollback_ansi_transaction_sql(depth))
                                     .map(|_| {
-                                        conn.transaction_depth -= 1;
+                                        shared.transaction_depth.fetch_sub(1, Ordering::Release);
                                     })
                             } else {
                                 Ok(())

sqlx-sqlite/src/transaction.rs

@@ -26,7 +26,7 @@ impl TransactionManager for SqliteTransactionManager {
         conn.worker.start_rollback().ok();
     }
 
-    fn get_transaction_depth(_conn: &SqliteConnection) -> usize {
-        todo!()
+    fn get_transaction_depth(conn: &SqliteConnection) -> usize {
+        conn.worker.shared.get_transaction_depth()
     }
 }