store: Replace uses of Pool.get_sync with Pool.get

Author: lutter

core/graphman/src/commands/deployment/info.rs

@@ -31,7 +31,7 @@ pub async fn load_deployments(
     deployment: &DeploymentSelector,
     version: &DeploymentVersionSelector,
 ) -> Result<Vec<Deployment>, GraphmanError> {
-    let mut primary_conn = primary_pool.get_sync().await?;
+    let mut primary_conn = primary_pool.get().await?;
 
     crate::deployment::load_deployments(&mut primary_conn, &deployment, &version).await
 }

store/postgres/src/graphman/mod.rs

@@ -2,6 +2,7 @@ use anyhow::Result;
 use async_trait::async_trait;
 use chrono::Utc;
 use diesel::prelude::*;
+use diesel_async::RunQueryDsl;
 use graphman_store::CommandKind;
 use graphman_store::Execution;
 use graphman_store::ExecutionId;
@@ -27,7 +28,7 @@ impl GraphmanStore {
 #[async_trait]
 impl graphman_store::GraphmanStore for GraphmanStore {
     async fn new_execution(&self, kind: CommandKind) -> Result<ExecutionId> {
-        let mut conn = self.primary_pool.get_sync().await?;
+        let mut conn = self.primary_pool.get().await?;
 
         let id: i64 = diesel::insert_into(gce::table)
             .values((
@@ -36,20 +37,21 @@ impl graphman_store::GraphmanStore for GraphmanStore {
                 gce::created_at.eq(Utc::now()),
             ))
             .returning(gce::id)
-            .get_result(&mut conn)?;
+            .get_result(&mut conn)
+            .await?;
 
         Ok(ExecutionId(id))
     }
 
     async fn load_execution(&self, id: ExecutionId) -> Result<Execution> {
-        let mut conn = self.primary_pool.get_sync().await?;
-        let execution = gce::table.find(id).first(&mut conn)?;
+        let mut conn = self.primary_pool.get().await?;
+        let execution = gce::table.find(id).first(&mut conn).await?;
 
         Ok(execution)
     }
 
     async fn mark_execution_as_running(&self, id: ExecutionId) -> Result<()> {
-        let mut conn = self.primary_pool.get_sync().await?;
+        let mut conn = self.primary_pool.get().await?;
 
         diesel::update(gce::table)
             .set((
@@ -58,13 +60,14 @@ impl graphman_store::GraphmanStore for GraphmanStore {
             ))
             .filter(gce::id.eq(id))
             .filter(gce::completed_at.is_null())
-            .execute(&mut conn)?;
+            .execute(&mut conn)
+            .await?;
 
         Ok(())
     }
 
     async fn mark_execution_as_failed(&self, id: ExecutionId, error_message: String) -> Result<()> {
-        let mut conn = self.primary_pool.get_sync().await?;
+        let mut conn = self.primary_pool.get().await?;
 
         diesel::update(gce::table)
             .set((
@@ -73,21 +76,23 @@ impl graphman_store::GraphmanStore for GraphmanStore {
                 gce::completed_at.eq(Utc::now()),
             ))
             .filter(gce::id.eq(id))
-            .execute(&mut conn)?;
+            .execute(&mut conn)
+            .await?;
 
         Ok(())
     }
 
     async fn mark_execution_as_succeeded(&self, id: ExecutionId) -> Result<()> {
-        let mut conn = self.primary_pool.get_sync().await?;
+        let mut conn = self.primary_pool.get().await?;
 
         diesel::update(gce::table)
             .set((
                 gce::status.eq(ExecutionStatus::Succeeded),
                 gce::completed_at.eq(Utc::now()),
             ))
             .filter(gce::id.eq(id))
-            .execute(&mut conn)?;
+            .execute(&mut conn)
+            .await?;
 
         Ok(())
     }

store/postgres/src/primary.rs

@@ -1991,7 +1991,7 @@ impl Primary {
     pub async fn is_source(&self, site: &Site) -> Result<bool, StoreError> {
         use active_copies as ac;
 
-        let mut conn = self.pool.get_sync().await?;
+        let mut conn = self.pool.get().await?;
 
         select(diesel::dsl::exists(
             ac::table
@@ -2006,7 +2006,7 @@ impl Primary {
     pub async fn is_copy_cancelled(&self, dst: &Site) -> Result<bool, StoreError> {
         use active_copies as ac;
 
-        let mut conn = self.pool.get_sync().await?;
+        let mut conn = self.pool.get().await?;
 
         ac::table
             .filter(ac::dst.eq(dst.id))

store/postgres/src/relational/prune.rs

@@ -1010,7 +1010,7 @@ mod status {
         pub async fn runs(&self) -> StoreResult<Vec<usize>> {
             use prune_state as ps;
 
-            let mut conn = self.pool.get_sync().await?;
+            let mut conn = self.pool.get().await?;
             let runs = ps::table
                 .filter(ps::id.eq(self.layout.site.id))
                 .select(ps::run)
@@ -1029,7 +1029,7 @@ mod status {
             use prune_state as ps;
             use prune_table_state as pts;
 
-            let mut conn = self.pool.get_sync().await?;
+            let mut conn = self.pool.get().await?;
 
             let ptss = pts::table
                 .filter(pts::id.eq(self.layout.site.id))

store/test-store/src/store.rs

@@ -410,15 +410,17 @@ pub async fn revert_block(store: &Arc<Store>, deployment: &DeploymentLocator, pt
 
 pub async fn insert_ens_name(hash: &str, name: &str) {
     use diesel::insert_into;
-    use diesel::prelude::*;
+    use diesel::ExpressionMethods;
+    use diesel_async::RunQueryDsl;
     use graph_store_postgres::command_support::catalog::ens_names;
 
-    let mut conn = PRIMARY_POOL.get_sync().await.unwrap();
+    let mut conn = PRIMARY_POOL.get().await.unwrap();
 
     insert_into(ens_names::table)
         .values((ens_names::hash.eq(hash), ens_names::name.eq(name)))
         .on_conflict_do_nothing()
         .execute(&mut conn)
+        .await
         .unwrap();
 }