store: Asyncify advisory_lock::lock

Author: lutter

store/postgres/src/advisory_lock.rs

@@ -53,7 +53,7 @@ impl Scope {
 
     /// Lock the deployment in this scope with the given id. Blocks until we
     /// can get the lock
-    fn lock(&self, conn: &mut PgConnection, id: DeploymentId) -> Result<(), StoreError> {
+    async fn lock(&self, conn: &mut PgConnection, id: DeploymentId) -> Result<(), StoreError> {
         sql_query(format!("select pg_advisory_lock({}, {id})", self.id))
             .execute(conn)
             .map(|_| ())
@@ -102,8 +102,8 @@ where
 
 /// Take the lock used to keep two copy operations to run simultaneously on
 /// the same deployment. Block until we can get the lock
-pub(crate) fn lock_copying(conn: &mut PgConnection, dst: &Site) -> Result<(), StoreError> {
-    COPY.lock(conn, dst.id)
+pub(crate) async fn lock_copying(conn: &mut PgConnection, dst: &Site) -> Result<(), StoreError> {
+    COPY.lock(conn, dst.id).await
 }
 
 /// Release the lock acquired with `lock_copying`.

store/postgres/src/copy.rs

@@ -701,12 +701,12 @@ impl LockTrackingConnection {
         }
     }
 
-    fn lock(&mut self, logger: &Logger, dst: &Site) -> Result<(), StoreError> {
+    async fn lock(&mut self, logger: &Logger, dst: &Site) -> Result<(), StoreError> {
         if self.has_lock {
             warn!(logger, "already acquired copy lock for {}", dst);
             return Ok(());
         }
-        advisory_lock::lock_copying(&mut self.inner, dst)?;
+        advisory_lock::lock_copying(&mut self.inner, dst).await?;
         self.has_lock = true;
         Ok(())
     }
@@ -1311,7 +1311,7 @@ impl Connection {
         let Some(conn) = self.conn.as_mut() else {
             return Err(internal_error!("copy connection went missing (copy_data)"));
         };
-        conn.lock(&self.logger, &dst_site)?;
+        conn.lock(&self.logger, &dst_site).await?;
 
         let res = self.copy_data_internal(index_list).await;