libsql: Use RwLock instead of RefCell for local connection (#2118)

The `local::Connection` struct has the following safety comments:

```rust
// SAFETY: This is safe because we compile sqlite3 w/ SQLITE_THREADSAFE=1
unsafe impl Send for Connection {}
// SAFETY: This is safe because we compile sqlite3 w/ SQLITE_THREADSAFE=1
unsafe impl Sync for Connection {}
```

This is correct with respect to the `*mut ffi::sqlite3` field, but NOT
with respect to the `RefCell<Option<AuthHook>>` field. RefCell is not
thread safe. Heavy load in a real application using local databases
occasionally sees panics due to concurrent access to this RefCell.

I don't have a simple repro right now, but this change eliminated the
panics for me. Let me know if you'd like some extra work done like a
repro or tests (not sure I'll have time but I will try..)

Author: penberg

libsql/Cargo.toml

@@ -63,6 +63,7 @@ core = [
   "dep:bitflags",
   "dep:bytes",
   "dep:futures",
+  "dep:parking_lot",
 ]
 stream = [
   "dep:futures",

libsql/src/local/connection.rs

@@ -12,8 +12,8 @@ use super::{Database, Error, Result, Rows, RowsFuture, Statement, Transaction};
 use crate::TransactionBehavior;
 
 use libsql_sys::ffi;
-use std::cell::RefCell;
 use std::{ffi::c_int, fmt, path::Path, sync::Arc};
+use parking_lot::RwLock;
 
 /// A connection to a libSQL database.
 #[derive(Clone)]
@@ -25,7 +25,7 @@ pub struct Connection {
     #[cfg(feature = "replication")]
     pub(crate) writer: Option<crate::replication::Writer>,
 
-    authorizer: RefCell<Option<AuthHook>>,
+    authorizer: Arc<RwLock<Option<AuthHook>>>,
 }
 
 impl Drop for Connection {
@@ -68,7 +68,7 @@ impl Connection {
             drop_ref: Arc::new(()),
             #[cfg(feature = "replication")]
             writer: db.writer()?,
-            authorizer: RefCell::new(None),
+            authorizer: Arc::new(RwLock::new(None)),
         };
         #[cfg(feature = "sync")]
         if let Some(_) = db.sync_ctx {
@@ -93,7 +93,7 @@ impl Connection {
             drop_ref: Arc::new(()),
             #[cfg(feature = "replication")]
             writer: None,
-            authorizer: RefCell::new(None),
+            authorizer: Arc::new(RwLock::new(None)),
         }
     }
 
@@ -473,7 +473,7 @@ impl Connection {
             }
         }
 
-        *self.authorizer.borrow_mut() = hook.clone();
+        *self.authorizer.write() = hook.clone();
 
         let (callback, user_data) = match hook {
             Some(_) => {
@@ -633,7 +633,7 @@ impl Connection {
     pub(crate) fn wal_insert_handle(&self) -> WalInsertHandle<'_> {
         WalInsertHandle {
             conn: self,
-            in_session: RefCell::new(false),
+            in_session: RwLock::new(false),
         }
     }
 }
@@ -647,7 +647,7 @@ unsafe extern "C" fn authorizer_callback(
     accessor: *const ::std::os::raw::c_char,
 ) -> ::std::os::raw::c_int {
     let conn = user_data as *const Connection;
-    let hook = unsafe { (*conn).authorizer.borrow() };
+    let hook = unsafe { (*conn).authorizer.read() };
     let hook = match &*hook {
         Some(hook) => hook,
         None => return ffi::SQLITE_OK,
@@ -688,37 +688,37 @@ unsafe extern "C" fn authorizer_callback(
 
 pub(crate) struct WalInsertHandle<'a> {
     conn: &'a Connection,
-    in_session: RefCell<bool>,
+    in_session: RwLock<bool>
 }
 
 impl WalInsertHandle<'_> {
     pub fn insert_at(&self, frame_no: u32, frame: &[u8]) -> Result<()> {
-        assert!(*self.in_session.borrow());
+        assert!(*self.in_session.read());
         self.conn.wal_insert_frame(frame_no, frame)
     }
 
     pub fn in_session(&self) -> bool {
-        *self.in_session.borrow()
+        *self.in_session.read()
     }
 
     pub fn begin(&self) -> Result<()> {
-        assert!(!*self.in_session.borrow());
+        assert!(!*self.in_session.read());
         self.conn.wal_insert_begin()?;
-        self.in_session.replace(true);
+        *self.in_session.write() = true;
         Ok(())
     }
 
     pub fn end(&self) -> Result<()> {
-        assert!(*self.in_session.borrow());
+        assert!(*self.in_session.read());
         self.conn.wal_insert_end()?;
-        self.in_session.replace(false);
+        *self.in_session.write() = false;
         Ok(())
     }
 }
 
 impl Drop for WalInsertHandle<'_> {
     fn drop(&mut self) {
-        if *self.in_session.borrow() {
+        if *self.in_session.read() {
             if let Err(err) = self.conn.wal_insert_end() {
                 tracing::error!("{:?}", err);
                 Err(err).unwrap()