fix(kv) run sqlite transactions via spawn_blocking (#19350)

`rusqlite` does not support async operations; with this PR SQLite
operations will run through `spawn_blocking` to ensure that the event
loop does not get blocked.

There is still only a single SQLite connection. So all operations will
do an async wait on the connection. In the future we can add a
connection pool if needed.

ext/kv/sqlite.rs

@@ -1,6 +1,7 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 
 use std::borrow::Cow;
+use std::cell::Cell;
 use std::cell::RefCell;
 use std::marker::PhantomData;
 use std::path::Path;
@@ -10,6 +11,8 @@ use std::rc::Rc;
 use async_trait::async_trait;
 use deno_core::error::type_error;
 use deno_core::error::AnyError;
+use deno_core::task::spawn_blocking;
+use deno_core::AsyncRefCell;
 use deno_core::OpState;
 use rusqlite::params;
 use rusqlite::OpenFlags;
@@ -112,11 +115,9 @@ impl<P: SqliteDbHandlerPermissions> DatabaseHandler for SqliteDbHandler<P> {
     state: Rc<RefCell<OpState>>,
     path: Option<String>,
   ) -> Result<Self::DB, AnyError> {
-    let conn = match (path.as_deref(), &self.default_storage_dir) {
-      (Some(":memory:"), _) | (None, None) => {
-        rusqlite::Connection::open_in_memory()?
-      }
-      (Some(path), _) => {
+    // Validate path
+    if let Some(path) = &path {
+      if path != ":memory:" {
         if path.is_empty() {
           return Err(type_error("Filename cannot be empty"));
         }
@@ -132,7 +133,18 @@ impl<P: SqliteDbHandlerPermissions> DatabaseHandler for SqliteDbHandler<P> {
           permissions.check_read(path, "Deno.openKv")?;
           permissions.check_write(path, "Deno.openKv")?;
         }
-        let flags = OpenFlags::default().difference(OpenFlags::SQLITE_OPEN_URI);
+      }
+    }
+
+    let default_storage_dir = self.default_storage_dir.clone();
+    let conn = spawn_blocking(move || {
+      let conn = match (path.as_deref(), &default_storage_dir) {
+        (Some(":memory:"), _) | (None, None) => {
+          rusqlite::Connection::open_in_memory()?
+        }
+        (Some(path), _) => {
+          let flags =
+            OpenFlags::default().difference(OpenFlags::SQLITE_OPEN_URI);
           rusqlite::Connection::open_with_flags(path, flags)?
         }
         (None, Some(path)) => {
@@ -165,11 +177,48 @@ impl<P: SqliteDbHandlerPermissions> DatabaseHandler for SqliteDbHandler<P> {
         }
       }
 
-    Ok(SqliteDb(RefCell::new(conn)))
+      Ok::<_, AnyError>(conn)
+    })
+    .await
+    .unwrap()?;
+
+    Ok(SqliteDb(Rc::new(AsyncRefCell::new(Cell::new(Some(conn))))))
   }
 }
 
-pub struct SqliteDb(RefCell<rusqlite::Connection>);
+pub struct SqliteDb(Rc<AsyncRefCell<Cell<Option<rusqlite::Connection>>>>);
+
+impl SqliteDb {
+  async fn run_tx<F, R>(&self, f: F) -> Result<R, AnyError>
+  where
+    F: (FnOnce(rusqlite::Transaction<'_>) -> Result<R, AnyError>)
+      + Send
+      + 'static,
+    R: Send + 'static,
+  {
+    // Transactions need exclusive access to the connection. Wait until
+    // we can borrow_mut the connection.
+    let cell = self.0.borrow_mut().await;
+
+    // Take the db out of the cell and run the transaction via spawn_blocking.
+    let mut db = cell.take().unwrap();
+    let (result, db) = spawn_blocking(move || {
+      let result = {
+        match db.transaction() {
+          Ok(tx) => f(tx),
+          Err(e) => Err(e.into()),
+        }
+      };
+      (result, db)
+    })
+    .await
+    .unwrap();
+
+    // Put the db back into the cell.
+    cell.set(Some(db));
+    result
+  }
+}
 
 #[async_trait(?Send)]
 impl Database for SqliteDb {
@@ -178,10 +227,9 @@ impl Database for SqliteDb {
     requests: Vec<ReadRange>,
     _options: SnapshotReadOptions,
   ) -> Result<Vec<ReadRangeOutput>, AnyError> {
+    self
+      .run_tx(move |tx| {
         let mut responses = Vec::with_capacity(requests.len());
-    let mut db = self.0.borrow_mut();
-    let tx = db.transaction()?;
-
         for request in requests {
           let mut stmt = tx.prepare_cached(if request.reverse {
             STATEMENT_KV_RANGE_SCAN_REVERSE
@@ -215,16 +263,16 @@ impl Database for SqliteDb {
         }
 
         Ok(responses)
+      })
+      .await
   }
 
   async fn atomic_write(
     &self,
     write: AtomicWrite,
   ) -> Result<Option<CommitResult>, AnyError> {
-    let mut db = self.0.borrow_mut();
-
-    let tx = db.transaction()?;
-
+    self
+      .run_tx(move |tx| {
         for check in write.checks {
           let real_versionstamp = tx
             .prepare_cached(STATEMENT_KV_POINT_GET_VERSION_ONLY)?
@@ -256,19 +304,34 @@ impl Database for SqliteDb {
               assert!(changed == 0 || changed == 1)
             }
             MutationKind::Sum(operand) => {
-          mutate_le64(&tx, &mutation.key, "sum", &operand, version, |a, b| {
-            a.wrapping_add(b)
-          })?;
+              mutate_le64(
+                &tx,
+                &mutation.key,
+                "sum",
+                &operand,
+                version,
+                |a, b| a.wrapping_add(b),
+              )?;
             }
             MutationKind::Min(operand) => {
-          mutate_le64(&tx, &mutation.key, "min", &operand, version, |a, b| {
-            a.min(b)
-          })?;
+              mutate_le64(
+                &tx,
+                &mutation.key,
+                "min",
+                &operand,
+                version,
+                |a, b| a.min(b),
+              )?;
             }
             MutationKind::Max(operand) => {
-          mutate_le64(&tx, &mutation.key, "max", &operand, version, |a, b| {
-            a.max(b)
-          })?;
+              mutate_le64(
+                &tx,
+                &mutation.key,
+                "max",
+                &operand,
+                version,
+                |a, b| a.max(b),
+              )?;
             }
           }
         }
@@ -282,6 +345,8 @@ impl Database for SqliteDb {
         Ok(Some(CommitResult {
           versionstamp: new_vesionstamp,
         }))
+      })
+      .await
   }
 }