server, store: Replace remaining uses of Pool.get_sync with Pool.get

Author: lutter

Cargo.lock

@@ -21,6 +21,7 @@ tower-http = { workspace = true }
 
 [dev-dependencies]
 diesel = { workspace = true }
+diesel-async = { workspace = true }
 lazy_static = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }

server/graphman/Cargo.toml

@@ -36,11 +36,12 @@ where
 }
 
 async fn cleanup_graphman_command_executions_table() {
-    use diesel::prelude::*;
+    use diesel_async::RunQueryDsl;
 
-    let mut conn = PRIMARY_POOL.get_sync().await.unwrap();
+    let mut conn = PRIMARY_POOL.get().await.unwrap();
 
     diesel::sql_query("truncate table public.graphman_command_executions;")
         .execute(&mut conn)
+        .await
         .expect("truncate is successful");
 }

server/graphman/tests/util/mod.rs

@@ -164,7 +164,7 @@ pub mod primary {
     }
 
     pub(super) async fn drop_chain(pool: &ConnectionPool, name: &str) -> Result<(), StoreError> {
-        let mut conn = pool.get_sync().await?;
+        let mut conn = pool.get().await?;
 
         delete(chains::table.filter(chains::name.eq(name)))
             .execute(&mut conn)
@@ -427,7 +427,7 @@ impl BlockStore {
             let cached = match self.chain_head_cache.get(shard.as_str()) {
                 Some(cached) => cached,
                 None => {
-                    let mut conn = match pool.get_sync().await {
+                    let mut conn = match pool.get().await {
                         Ok(conn) => conn,
                         Err(StoreError::DatabaseUnavailable) => continue,
                         Err(e) => return Err(e),
@@ -608,7 +608,7 @@ impl BlockStore {
             None => {}
         }
 
-        let mut conn = self.mirror.primary().get_sync().await?;
+        let mut conn = self.mirror.primary().get().await?;
         let shard = self
             .shards
             .iter()
@@ -667,7 +667,7 @@ impl ChainIdStore for BlockStore {
 
         // Update the master copy in the primary
         let primary_pool = self.pools.get(&*PRIMARY_SHARD).unwrap();
-        let mut conn = primary_pool.get_sync().await?;
+        let mut conn = primary_pool.get().await?;
 
         diesel::update(c::table.filter(c::name.eq(chain_name.as_str())))
             .set((

store/postgres/src/block_store.rs

@@ -2131,7 +2131,7 @@ impl ChainStore {
     pub(crate) async fn set_chain_identifier(&self, ident: &ChainIdentifier) -> Result<(), Error> {
         use public::ethereum_networks as n;
 
-        let mut conn = self.pool.get_sync().await?;
+        let mut conn = self.pool.get().await?;
 
         diesel::update(n::table.filter(n::name.eq(&self.chain)))
             .set((
@@ -2163,7 +2163,7 @@ impl ChainStore {
     ) -> Vec<(BlockPtr, BlockHash)> {
         let mut conn = self
             .pool
-            .get_sync()
+            .get()
             .await
             .expect("can get a database connection");
 
@@ -2440,7 +2440,7 @@ impl ChainStoreTrait for ChainStore {
     }
 
     async fn upsert_light_blocks(&self, blocks: &[&dyn Block]) -> Result<(), Error> {
-        let mut conn = self.pool.get_sync().await?;
+        let mut conn = self.pool.get().await?;
         for block in blocks {
             self.storage
                 .upsert_block(&mut conn, &self.chain, *block, false)
@@ -2795,7 +2795,7 @@ impl ChainStoreTrait for ChainStore {
     }
 
     async fn chain_identifier(&self) -> Result<ChainIdentifier, Error> {
-        let mut conn = self.pool.get_sync().await?;
+        let mut conn = self.pool.get().await?;
         use public::ethereum_networks as n;
         let (genesis_block_hash, net_version) = n::table
             .select((n::genesis_block_hash, n::net_version))

store/postgres/src/chain_store.rs

@@ -546,7 +546,7 @@ fn test_clear_stale_call_cache() {
         let call: [u8; 6] = [1, 2, 3, 4, 5, 6];
         let return_value: [u8; 3] = [7, 8, 9];
 
-        let mut conn = PRIMARY_POOL.get_sync().await.unwrap();
+        let mut conn = PRIMARY_POOL.get().await.unwrap();
 
         // Insert a call cache entry, otherwise it will hit an early return and won't test all queries
         let call = call::Request::new(address, call.to_vec(), 0);