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

Author: lutter

node/src/manager/commands/chain.rs

@@ -1,7 +1,8 @@
 use std::sync::Arc;
 
 use diesel::sql_query;
-use diesel::RunQueryDsl;
+use diesel_async::AsyncConnection;
+use diesel_async::RunQueryDsl;
 use graph::blockchain::BlockHash;
 use graph::blockchain::BlockPtr;
 use graph::blockchain::ChainIdentifier;
@@ -24,7 +25,6 @@ use graph_chain_ethereum::EthereumAdapterTrait as _;
 use graph_store_postgres::add_chain;
 use graph_store_postgres::find_chain;
 use graph_store_postgres::update_chain_name;
-use graph_store_postgres::AsyncConnection;
 use graph_store_postgres::BlockStore;
 use graph_store_postgres::ChainStatus;
 use graph_store_postgres::ChainStore;
@@ -37,7 +37,7 @@ use crate::network_setup::Networks;
 
 pub async fn list(primary: ConnectionPool, store: BlockStore) -> Result<(), Error> {
     let mut chains = {
-        let mut conn = primary.get_sync().await?;
+        let mut conn = primary.get().await?;
         block_store::load_chains(&mut conn).await?
     };
     chains.sort_by_key(|chain| chain.name.clone());
@@ -122,7 +122,7 @@ pub async fn info(
         }
     }
 
-    let mut conn = primary.get_sync().await?;
+    let mut conn = primary.get().await?;
 
     let chain = block_store::find_chain(&mut conn, &name)
         .await?
@@ -231,7 +231,7 @@ pub async fn change_block_cache_shard(
 ) -> Result<(), Error> {
     println!("Changing block cache shard for {} to {}", chain_name, shard);
 
-    let mut conn = primary_store.get_sync().await?;
+    let mut conn = primary_store.get().await?;
 
     let chain = find_chain(&mut conn, &chain_name)
         .await?
@@ -247,7 +247,7 @@ pub async fn change_block_cache_shard(
     let new_name = format!("{}-old", &chain_name);
     let ident = chain_store.chain_identifier().await?;
 
-    conn.transaction_async::<(), StoreError, _>(|conn|  {
+    conn.transaction::<(), StoreError, _>(|conn|  {
         async {
             let shard = Shard::new(shard.to_string())?;
 
@@ -259,7 +259,7 @@ pub async fn change_block_cache_shard(
             sql_query(
                 "alter table deployment_schemas drop constraint deployment_schemas_network_fkey;",
             )
-            .execute(conn)?;
+            .execute(conn).await?;
 
             // Update the current chain name to chain-old
             update_chain_name(conn, &chain_name, &new_name).await?;
@@ -271,7 +271,7 @@ pub async fn change_block_cache_shard(
             sql_query(
                 "alter table deployment_schemas add constraint deployment_schemas_network_fkey foreign key (network) references chains(name);",
             )
-            .execute(conn)?;
+            .execute(conn).await?;
             Ok(())
         }.scope_boxed()
     }).await?;

node/src/manager/commands/copy.rs

@@ -1,4 +1,5 @@
-use diesel::{ExpressionMethods, JoinOnDsl, OptionalExtension, QueryDsl, RunQueryDsl};
+use diesel::{ExpressionMethods, JoinOnDsl, OptionalExtension, QueryDsl};
+use diesel_async::RunQueryDsl;
 use std::{collections::HashMap, sync::Arc};
 
 use graph::{
@@ -67,18 +68,20 @@ impl CopyState {
             .get(shard)
             .ok_or_else(|| anyhow!("can not find pool for shard {}", shard))?;
 
-        let mut dconn = dpool.get_sync().await?;
+        let mut dconn = dpool.get().await?;
 
         let tables = cts::table
             .filter(cts::dst.eq(dst))
             .order_by(cts::entity_type)
-            .load::<CopyTableState>(&mut dconn)?;
+            .load::<CopyTableState>(&mut dconn)
+            .await?;
 
         let on_sync = on_sync(&mut dconn, DeploymentId(dst)).await?;
 
         Ok(cs::table
             .filter(cs::dst.eq(dst))
             .get_result::<CopyState>(&mut dconn)
+            .await
             .optional()?
             .map(|state| (state, tables, on_sync)))
     }
@@ -209,7 +212,7 @@ pub async fn list(pools: HashMap<Shard, ConnectionPool>) -> Result<(), Error> {
     use catalog::deployment_schemas as ds;
 
     let primary = pools.get(&*PRIMARY_SHARD).expect("there is a primary pool");
-    let mut conn = primary.get_sync().await?;
+    let mut conn = primary.get().await?;
 
     let copies = ac::table
         .inner_join(ds::table.on(ds::id.eq(ac::dst)))
@@ -221,7 +224,8 @@ pub async fn list(pools: HashMap<Shard, ConnectionPool>) -> Result<(), Error> {
             ds::subgraph,
             ds::shard,
         ))
-        .load::<(i32, i32, Option<UtcDateTime>, UtcDateTime, String, Shard)>(&mut conn)?;
+        .load::<(i32, i32, Option<UtcDateTime>, UtcDateTime, String, Shard)>(&mut conn)
+        .await?;
     if copies.is_empty() {
         println!("no active copies");
     } else {
@@ -274,18 +278,20 @@ pub async fn status(
     let primary = pools
         .get(&*PRIMARY_SHARD)
         .ok_or_else(|| anyhow!("can not find deployment with id {}", dst))?;
-    let mut pconn = primary.get_sync().await?;
+    let mut pconn = primary.get().await?;
     let dst = dst.locate_unique(primary).await?.id.0;
 
     let (shard, deployment) = ds::table
         .filter(ds::id.eq(dst))
         .select((ds::shard, ds::subgraph))
-        .get_result::<(Shard, String)>(&mut pconn)?;
+        .get_result::<(Shard, String)>(&mut pconn)
+        .await?;
 
     let (active, cancelled_at) = ac::table
         .filter(ac::dst.eq(dst))
         .select((ac::src, ac::cancelled_at))
         .get_result::<(i32, Option<UtcDateTime>)>(&mut pconn)
+        .await
         .optional()?
         .map(|(_, cancelled_at)| (true, cancelled_at))
         .unwrap_or((false, None));

node/src/manager/commands/stats.rs

@@ -10,16 +10,16 @@ use graph::prelude::anyhow;
 use graph::prelude::CheapClone as _;
 use graph_store_postgres::command_support::catalog as store_catalog;
 use graph_store_postgres::command_support::catalog::Site;
+use graph_store_postgres::AsyncPgConnection;
 use graph_store_postgres::ConnectionPool;
-use graph_store_postgres::PgConnection;
 use graph_store_postgres::Shard;
 use graph_store_postgres::SubgraphStore;
 use graph_store_postgres::PRIMARY_SHARD;
 
 async fn site_and_conn(
     pools: HashMap<Shard, ConnectionPool>,
     search: &DeploymentSearch,
-) -> Result<(Arc<Site>, PgConnection), anyhow::Error> {
+) -> Result<(Arc<Site>, AsyncPgConnection), anyhow::Error> {
     let primary_pool = pools.get(&*PRIMARY_SHARD).unwrap();
     let locator = search.locate_unique(primary_pool).await?;
 
@@ -32,7 +32,7 @@ async fn site_and_conn(
         .ok_or_else(|| anyhow!("deployment `{}` does not exist", search))?;
     let site = Arc::new(site);
 
-    let conn = pools.get(&site.shard).unwrap().get_sync().await?;
+    let conn = pools.get(&site.shard).unwrap().get().await?;
 
     Ok((site, conn))
 }

node/src/manager/commands/txn_speed.rs

@@ -32,7 +32,7 @@ pub async fn run(pool: ConnectionPool, delay: u64) -> Result<(), anyhow::Error>
     }
 
     let mut speeds = HashMap::new();
-    let mut conn = pool.get_sync().await?;
+    let mut conn = pool.get().await?;
     for (datname, all_txn, write_txn) in query(&mut conn).await? {
         speeds.insert(datname, (all_txn, write_txn));
     }

node/src/manager/deployment.rs

@@ -93,7 +93,7 @@ impl DeploymentSearch {
     }
 
     pub async fn lookup(&self, primary: &ConnectionPool) -> Result<Vec<Deployment>, anyhow::Error> {
-        let mut conn = primary.get_sync().await?;
+        let mut conn = primary.get().await?;
         self.lookup_with_conn(&mut conn).await
     }