store: Asyncify remaining methods in block_store

Also, remove unused update_chain_genesis_hash

Author: lutter

node/src/manager/commands/chain.rs

@@ -38,7 +38,7 @@ use crate::network_setup::Networks;
 pub async fn list(primary: ConnectionPool, store: BlockStore) -> Result<(), Error> {
     let mut chains = {
         let mut conn = primary.get_async().await?;
-        block_store::load_chains(&mut conn)?
+        block_store::load_chains(&mut conn).await?
     };
     chains.sort_by_key(|chain| chain.name.clone());
 
@@ -124,7 +124,8 @@ pub async fn info(
 
     let mut conn = primary.get_async().await?;
 
-    let chain = block_store::find_chain(&mut conn, &name)?
+    let chain = block_store::find_chain(&mut conn, &name)
+        .await?
         .ok_or_else(|| anyhow!("unknown chain: {}", name))?;
 
     let chain_store = store
@@ -233,7 +234,8 @@ pub async fn change_block_cache_shard(
 
     let mut conn = primary_store.get_async().await?;
 
-    let chain = find_chain(&mut conn, &chain_name)?
+    let chain = find_chain(&mut conn, &chain_name)
+        .await?
         .ok_or_else(|| anyhow!("unknown chain: {}", chain_name))?;
     let old_shard = chain.shard;
 
@@ -250,7 +252,7 @@ pub async fn change_block_cache_shard(
         async {
             let shard = Shard::new(shard.to_string())?;
 
-            let chain = BlockStore::allocate_chain(conn, &chain_name, &shard, &ident)?;
+            let chain = BlockStore::allocate_chain(conn, &chain_name, &shard, &ident).await?;
 
             graph::block_on(store.add_chain_store(&chain,ChainStatus::Ingestible, true))?;
 
@@ -261,10 +263,10 @@ pub async fn change_block_cache_shard(
             .execute(conn)?;
 
             // Update the current chain name to chain-old
-            update_chain_name(conn, &chain_name, &new_name)?;
+            update_chain_name(conn, &chain_name, &new_name).await?;
 
             // Create a new chain with the name in the destination shard
-            let _ = add_chain(conn, &chain_name, &shard, ident)?;
+            let _ = add_chain(conn, &chain_name, &shard, ident).await?;
 
             // Re-add the foreign key constraint
             sql_query(

store/postgres/src/block_store.rs

@@ -54,9 +54,8 @@ pub mod primary {
     use std::convert::TryFrom;
 
     use diesel::{
-        delete, insert_into,
-        r2d2::{ConnectionManager, PooledConnection},
-        update, ExpressionMethods, OptionalExtension, PgConnection, QueryDsl, RunQueryDsl,
+        delete, insert_into, update, ExpressionMethods, OptionalExtension, PgConnection, QueryDsl,
+        RunQueryDsl,
     };
     use graph::{
         blockchain::{BlockHash, ChainIdentifier},
@@ -109,18 +108,21 @@ pub mod primary {
         }
     }
 
-    pub fn load_chains(conn: &mut PgConnection) -> Result<Vec<Chain>, StoreError> {
+    pub async fn load_chains(conn: &mut PgConnection) -> Result<Vec<Chain>, StoreError> {
         Ok(chains::table.load(conn)?)
     }
 
-    pub fn find_chain(conn: &mut PgConnection, name: &str) -> Result<Option<Chain>, StoreError> {
+    pub async fn find_chain(
+        conn: &mut PgConnection,
+        name: &str,
+    ) -> Result<Option<Chain>, StoreError> {
         Ok(chains::table
             .filter(chains::name.eq(name))
             .first(conn)
             .optional()?)
     }
 
-    pub fn add_chain(
+    pub async fn add_chain(
         conn: &mut PgConnection,
         name: &str,
         shard: &Shard,
@@ -165,7 +167,7 @@ pub mod primary {
     }
 
     // update chain name where chain name is 'name'
-    pub fn update_chain_name(
+    pub async fn update_chain_name(
         conn: &mut PgConnection,
         name: &str,
         new_name: &str,
@@ -175,17 +177,6 @@ pub mod primary {
             .execute(conn)?;
         Ok(())
     }
-
-    pub fn update_chain_genesis_hash(
-        conn: &mut PooledConnection<ConnectionManager<PgConnection>>,
-        name: &str,
-        hash: BlockHash,
-    ) -> Result<(), StoreError> {
-        update(chains::table.filter(chains::name.eq(name)))
-            .set(chains::genesis_block_hash.eq(hash.hash_hex()))
-            .execute(conn)?;
-        Ok(())
-    }
 }
 
 /// The store that chains use to maintain their state and cache often used
@@ -265,7 +256,7 @@ impl BlockStore {
 
         let mirror = PrimaryMirror::new(&pools);
         let existing_chains = mirror
-            .read_async(|conn| async { primary::load_chains(conn) }.scope_boxed())
+            .read_async(|conn| primary::load_chains(conn).scope_boxed())
             .await?;
         let chain_head_cache = TimedCache::new(CHAIN_HEAD_CACHE_TTL);
         let chains = shards.clone();
@@ -345,7 +336,7 @@ impl BlockStore {
         self.mirror.primary().query_permit().await
     }
 
-    pub fn allocate_chain(
+    pub async fn allocate_chain(
         conn: &mut PgConnection,
         name: &String,
         shard: &Shard,
@@ -463,7 +454,7 @@ impl BlockStore {
         self.mirror
             .read_async(|conn| {
                 async {
-                    match primary::find_chain(conn, &chain)? {
+                    match primary::find_chain(conn, &chain).await? {
                         Some(chain) => {
                             let chain_store = this
                                 .add_chain_store(&chain, ChainStatus::ReadOnly, false)
@@ -617,7 +608,7 @@ impl BlockStore {
                 }
             })
             .ok_or_else(|| anyhow!("unable to find shard for network {}", network))?;
-        let chain = primary::add_chain(&mut conn, &network, &shard, ident)?;
+        let chain = primary::add_chain(&mut conn, &network, &shard, ident).await?;
         self.add_chain_store(&chain, ChainStatus::Ingestible, true)
             .await
             .map_err(anyhow::Error::from)