feat: add async to delete_all (#621)

* elevated `begin()` to an interface call to allow `delete_all` the
ability to lock the database.
* added `begin_async()` wrapper to mysql & mock to mirror spanner function
* added `begin()` call to `delete_all`

Closes #615

Co-authored-by: Donovan Preston <donovanpreston@gmail.com>

src/db/mock.rs

@@ -57,6 +57,10 @@ impl Db for MockDb {
         Box::pin(future::ok(()))
     }
 
+    fn begin(&self, _for_write: bool) -> DbFuture<()> {
+        Box::pin(future::ok(()))
+    }
+
     fn box_clone(&self) -> Box<dyn Db> {
         Box::new(self.clone())
     }

src/db/mod.rs

@@ -79,6 +79,8 @@ pub trait Db: Send + Debug {
 
     fn lock_for_write(&self, params: params::LockCollection) -> DbFuture<()>;
 
+    fn begin(&self, for_write: bool) -> DbFuture<()>;
+
     fn commit(&self) -> DbFuture<()>;
 
     fn rollback(&self) -> DbFuture<()>;

src/db/mysql/models.rs

@@ -236,6 +236,10 @@ impl MysqlDb {
         Ok(())
     }
 
+    pub async fn begin_async(&self, for_write: bool) -> Result<()> {
+        self.begin(for_write)
+    }
+
     pub fn commit_sync(&self) -> Result<()> {
         if self.session.borrow().in_transaction {
             self.conn
@@ -911,6 +915,11 @@ impl Db for MysqlDb {
         Box::pin(block(move || db.rollback_sync().map_err(Into::into)).map_err(Into::into))
     }
 
+    fn begin(&self, for_write: bool) -> DbFuture<()> {
+        let db = self.clone();
+        Box::pin(async move { db.begin_async(for_write).map_err(Into::into).await })
+    }
+
     fn box_clone(&self) -> Box<dyn Db> {
         Box::new(self.clone())
     }

src/db/spanner/models.rs

@@ -1639,6 +1639,11 @@ impl Db for SpannerDb {
         Box::pin(async move { db.lock_for_write_async(param).map_err(Into::into).await })
     }
 
+    fn begin(&self, for_write: bool) -> DbFuture<()> {
+        let db = self.clone();
+        Box::pin(async move { db.begin_async(for_write).map_err(Into::into).await })
+    }
+
     fn get_collection_timestamp(
         &self,
         param: params::GetCollectionTimestamp,

src/web/handlers.rs

@@ -69,7 +69,12 @@ pub async fn get_quota(meta: MetaRequest) -> Result<HttpResponse, Error> {
 pub async fn delete_all(meta: MetaRequest) -> Result<HttpResponse, Error> {
     #![allow(clippy::unit_arg)]
     meta.metrics.incr("request.delete_all");
-    Ok(HttpResponse::Ok().json(meta.db.delete_storage(meta.user_id).await?))
+    let db = meta.db;
+    db.begin(true).await?;
+    match db.delete_storage(meta.user_id).await {
+        Ok(r) => Ok(HttpResponse::Ok().json(r)),
+        Err(e) => Err(e.into()),
+    }
 }
 
 pub fn delete_collection(