store: Asyncify remaining methods in copy

Author: lutter

store/postgres/src/copy.rs

@@ -297,7 +297,7 @@ impl CopyState {
     }
 }
 
-pub(crate) fn source(
+pub(crate) async fn source(
     conn: &mut PgConnection,
     dst: &Site,
 ) -> Result<Option<DeploymentId>, StoreError> {
@@ -429,7 +429,7 @@ impl TableState {
         Ok(states)
     }
 
-    fn record_progress(
+    async fn record_progress(
         &mut self,
         conn: &mut PgConnection,
         elapsed: Duration,
@@ -466,7 +466,7 @@ impl TableState {
         Ok(())
     }
 
-    fn record_finished(&self, conn: &mut PgConnection) -> Result<(), StoreError> {
+    async fn record_finished(&self, conn: &mut PgConnection) -> Result<(), StoreError> {
         use copy_table_state as cts;
 
         update(
@@ -509,16 +509,20 @@ impl TableState {
 
         deployment::update_entity_count(conn, &self.dst_site, count).await?;
 
-        self.record_progress(conn, duration)?;
+        self.record_progress(conn, duration).await?;
 
         if self.finished() {
-            self.record_finished(conn)?;
+            self.record_finished(conn).await?;
         }
 
         Ok(Status::Finished)
     }
 
-    fn set_batch_size(&mut self, conn: &mut PgConnection, size: usize) -> Result<(), StoreError> {
+    async fn set_batch_size(
+        &mut self,
+        conn: &mut PgConnection,
+        size: usize,
+    ) -> Result<(), StoreError> {
         use copy_table_state as cts;
 
         self.batcher.set_batch_size(size);
@@ -835,10 +839,8 @@ impl CopyTableWorker {
                     // that is hard to predict. This mechanism ensures
                     // that if our estimation is wrong, the consequences
                     // aren't too severe.
-                    conn.transaction_async(|conn| {
-                        async { self.table.set_batch_size(conn, 1) }.scope_boxed()
-                    })
-                    .await?;
+                    conn.transaction_async(|conn| self.table.set_batch_size(conn, 1).scope_boxed())
+                        .await?;
                 }
             };
 

store/postgres/src/deployment_store.rs

@@ -595,9 +595,12 @@ impl DeploymentStore {
 
     /// Return the source if `site` or `None` if `site` is neither a graft
     /// nor a copy
-    pub(crate) fn source_of_copy(&self, site: &Site) -> Result<Option<DeploymentId>, StoreError> {
+    pub(crate) async fn source_of_copy(
+        &self,
+        site: &Site,
+    ) -> Result<Option<DeploymentId>, StoreError> {
         let mut conn = self.get_conn()?;
-        crate::copy::source(&mut conn, site)
+        crate::copy::source(&mut conn, site).await
     }
 
     // Only used for tests

store/postgres/src/writable.rs

@@ -464,7 +464,7 @@ impl SyncStore {
             // Handle on_sync actions. They only apply to copies (not
             // grafts) so we make sure that the source, if it exists, has
             // the same hash as `self.site`
-            if let Some(src) = self.writable.source_of_copy(&self.site)? {
+            if let Some(src) = self.writable.source_of_copy(&self.site).await? {
                 if let Some(src) = self.maybe_find_site(src).await? {
                     if src.deployment == self.site.deployment {
                         let on_sync = self.writable.on_sync(&self.site).await?;