store: Asyncify advisory_lock::unlock

Author: lutter

store/postgres/src/advisory_lock.rs

@@ -61,7 +61,7 @@ impl Scope {
     }
 
     /// Unlock the deployment in this scope with the given id.
-    fn unlock(&self, conn: &mut PgConnection, id: DeploymentId) -> Result<(), StoreError> {
+    async fn unlock(&self, conn: &mut PgConnection, id: DeploymentId) -> Result<(), StoreError> {
         sql_query(format!("select pg_advisory_unlock({}, {id})", self.id))
             .execute(conn)
             .map(|_| ())
@@ -107,8 +107,8 @@ pub(crate) async fn lock_copying(conn: &mut PgConnection, dst: &Site) -> Result<
 }
 
 /// Release the lock acquired with `lock_copying`.
-pub(crate) fn unlock_copying(conn: &mut PgConnection, dst: &Site) -> Result<(), StoreError> {
-    COPY.unlock(conn, dst.id)
+pub(crate) async fn unlock_copying(conn: &mut PgConnection, dst: &Site) -> Result<(), StoreError> {
+    COPY.unlock(conn, dst.id).await
 }
 
 /// Take the lock used to keep two operations from writing to the deployment
@@ -123,11 +123,11 @@ pub(crate) async fn lock_deployment_session(
 }
 
 /// Release the lock acquired with `lock_deployment_session`.
-pub(crate) fn unlock_deployment_session(
+pub(crate) async fn unlock_deployment_session(
     conn: &mut PgConnection,
     site: &Site,
 ) -> Result<(), StoreError> {
-    WRITE.unlock(conn, site.id)
+    WRITE.unlock(conn, site.id).await
 }
 
 /// Try to take the lock used to prevent two prune operations from running at the
@@ -139,6 +139,6 @@ pub(crate) async fn try_lock_pruning(
     PRUNE.try_lock(conn, site.id).await
 }
 
-pub(crate) fn unlock_pruning(conn: &mut PgConnection, site: &Site) -> Result<(), StoreError> {
-    PRUNE.unlock(conn, site.id)
+pub(crate) async fn unlock_pruning(conn: &mut PgConnection, site: &Site) -> Result<(), StoreError> {
+    PRUNE.unlock(conn, site.id).await
 }

store/postgres/src/copy.rs

@@ -711,15 +711,15 @@ impl LockTrackingConnection {
         Ok(())
     }
 
-    fn unlock(&mut self, logger: &Logger, dst: &Site) -> Result<(), StoreError> {
+    async fn unlock(&mut self, logger: &Logger, dst: &Site) -> Result<(), StoreError> {
         if !self.has_lock {
             error!(
                 logger,
                 "tried to release copy lock for {} even though we are not the owner", dst
             );
             return Ok(());
         }
-        advisory_lock::unlock_copying(&mut self.inner, dst)?;
+        advisory_lock::unlock_copying(&mut self.inner, dst).await?;
         self.has_lock = false;
         Ok(())
     }
@@ -1327,7 +1327,7 @@ impl Connection {
                 );
             }
             Some(conn) => {
-                conn.unlock(&self.logger, &dst_site)?;
+                conn.unlock(&self.logger, &dst_site).await?;
             }
         }
 

store/postgres/src/deployment.rs

@@ -1396,6 +1396,6 @@ where
         backoff.sleep();
     }
     let res = f(conn).await;
-    advisory_lock::unlock_deployment_session(conn, site)?;
+    advisory_lock::unlock_deployment_session(conn, site).await?;
     res
 }

store/postgres/src/deployment_store.rs

@@ -896,7 +896,7 @@ impl DeploymentStore {
             // kick off a pruning run while this node might still be pruning
             if advisory_lock::try_lock_pruning(conn, &site).await? {
                 let res = do_prune(store, conn, site.cheap_clone(), cancel, req, reporter).await;
-                advisory_lock::unlock_pruning(conn, &site)?;
+                advisory_lock::unlock_pruning(conn, &site).await?;
                 res
             } else {
                 Ok(reporter)