node: Use AsyncPgConnection in remaining places

Author: lutter

Cargo.lock

@@ -37,6 +37,7 @@ serde = { workspace = true }
 shellexpand = "3.1.1"
 termcolor = "1.4.1"
 diesel = { workspace = true }
+diesel-async = { workspace = true }
 prometheus = { version = "0.14.0", features = ["push"] }
 json-structural-diff = { version = "0.2", features = ["colorize"] }
 globset = "0.4.16"

node/Cargo.toml

@@ -1,16 +1,17 @@
 use std::{collections::HashMap, thread::sleep, time::Duration};
 
+use diesel::dsl::sql;
 use graph::prelude::anyhow;
-use graph_store_postgres::{ConnectionPool, PgConnection};
+use graph_store_postgres::{AsyncPgConnection, ConnectionPool};
 
 use crate::manager::catalog;
 
 pub async fn run(pool: ConnectionPool, delay: u64) -> Result<(), anyhow::Error> {
-    fn query(conn: &mut PgConnection) -> Result<Vec<(String, i64, i64)>, anyhow::Error> {
+    async fn query(conn: &mut AsyncPgConnection) -> Result<Vec<(String, i64, i64)>, anyhow::Error> {
         use catalog::pg_catalog::pg_stat_database as d;
-        use diesel::dsl::*;
         use diesel::sql_types::BigInt;
-        use diesel::{ExpressionMethods, QueryDsl, RunQueryDsl};
+        use diesel::{ExpressionMethods, QueryDsl};
+        use diesel_async::RunQueryDsl;
 
         let rows = d::table
             .filter(d::datname.eq_any(vec!["explorer", "graph"]))
@@ -20,7 +21,8 @@ pub async fn run(pool: ConnectionPool, delay: u64) -> Result<(), anyhow::Error>
                 sql::<BigInt>("txid_current()::bigint"),
             ))
             //.select((d::datname))
-            .load::<(Option<String>, i64, i64)>(conn)?;
+            .load::<(Option<String>, i64, i64)>(conn)
+            .await?;
         Ok(rows
             .into_iter()
             .map(|(datname, all_txn, write_txn)| {
@@ -31,7 +33,7 @@ pub async fn run(pool: ConnectionPool, delay: u64) -> Result<(), anyhow::Error>
 
     let mut speeds = HashMap::new();
     let mut conn = pool.get_sync().await?;
-    for (datname, all_txn, write_txn) in query(&mut conn)? {
+    for (datname, all_txn, write_txn) in query(&mut conn).await? {
         speeds.insert(datname, (all_txn, write_txn));
     }
     println!(
@@ -41,7 +43,7 @@ pub async fn run(pool: ConnectionPool, delay: u64) -> Result<(), anyhow::Error>
     sleep(Duration::from_secs(delay));
     println!("Number of transactions/minute");
     println!("{:10} {:>7} write", "database", "all");
-    for (datname, all_txn, write_txn) in query(&mut conn)? {
+    for (datname, all_txn, write_txn) in query(&mut conn).await? {
         let (all_speed, write_speed) = speeds
             .get(&datname)
             .map(|(all_txn_old, write_txn_old)| {

node/src/manager/commands/txn_speed.rs

@@ -6,17 +6,17 @@ use diesel::dsl::sql;
 use diesel::sql_types::Text;
 use diesel::{
     ExpressionMethods, JoinOnDsl, NullableExpressionMethods, PgTextExpressionMethods, QueryDsl,
-    RunQueryDsl,
 };
+use diesel_async::RunQueryDsl;
 
 use graph::components::store::DeploymentId;
 use graph::{
     components::store::DeploymentLocator,
     prelude::{anyhow, lazy_static, regex::Regex, DeploymentHash},
 };
 use graph_store_postgres::command_support::catalog as store_catalog;
-use graph_store_postgres::ConnectionPool;
-use graph_store_postgres::{unused, PgConnection};
+use graph_store_postgres::unused;
+use graph_store_postgres::{AsyncPgConnection, ConnectionPool};
 
 lazy_static! {
     // `Qm...` optionally follow by `:$shard`
@@ -94,12 +94,12 @@ impl DeploymentSearch {
 
     pub async fn lookup(&self, primary: &ConnectionPool) -> Result<Vec<Deployment>, anyhow::Error> {
         let mut conn = primary.get_sync().await?;
-        self.lookup_with_conn(&mut conn)
+        self.lookup_with_conn(&mut conn).await
     }
 
-    pub fn lookup_with_conn(
+    pub async fn lookup_with_conn(
         &self,
-        conn: &mut PgConnection,
+        conn: &mut AsyncPgConnection,
     ) -> Result<Vec<Deployment>, anyhow::Error> {
         use store_catalog::deployment_schemas as ds;
         use store_catalog::subgraph as s;
@@ -130,19 +130,19 @@ impl DeploymentSearch {
         let deployments: Vec<Deployment> = match self {
             DeploymentSearch::Name { name } => {
                 let pattern = format!("%{}%", name);
-                query.filter(s::name.ilike(&pattern)).load(conn)?
+                query.filter(s::name.ilike(&pattern)).load(conn).await?
             }
             DeploymentSearch::Hash { hash, shard } => {
                 let query = query.filter(ds::subgraph.eq(&hash));
                 match shard {
-                    Some(shard) => query.filter(ds::shard.eq(shard)).load(conn)?,
-                    None => query.load(conn)?,
+                    Some(shard) => query.filter(ds::shard.eq(shard)).load(conn).await?,
+                    None => query.load(conn).await?,
                 }
             }
             DeploymentSearch::Deployment { namespace } => {
-                query.filter(ds::name.eq(&namespace)).load(conn)?
+                query.filter(ds::name.eq(&namespace)).load(conn).await?
             }
-            DeploymentSearch::All => query.load(conn)?,
+            DeploymentSearch::All => query.load(conn).await?,
         };
         Ok(deployments)
     }