store: Use an async connection pool

We use an async connnection pool, but only with a synchronous wrapper so
that most uses of connections are not affected by the change yet. Over
time, we will change code to use AsyncPgConn and make it truly async.

Unfortunatley, this change also involves a lot of line noise because of the
replacement of diesel's PgConnection with our own type definition that uses
a AsyncConnectionWrapper

Author: lutter

core/graphman/src/deployment.rs

@@ -1,11 +1,19 @@
 use anyhow::anyhow;
 use diesel::dsl::sql;
-use diesel::prelude::*;
 use diesel::sql_types::Text;
+use diesel::BoolExpressionMethods;
+use diesel::ExpressionMethods;
+use diesel::JoinOnDsl;
+use diesel::NullableExpressionMethods;
+use diesel::PgTextExpressionMethods;
+use diesel::QueryDsl;
+use diesel::Queryable;
+use diesel::RunQueryDsl;
 use graph::components::store::DeploymentId;
 use graph::components::store::DeploymentLocator;
 use graph::data::subgraph::DeploymentHash;
 use graph_store_postgres::command_support::catalog;
+use graph_store_postgres::PgConnection;
 use itertools::Itertools;
 
 use crate::GraphmanError;

node/src/manager/commands/stats.rs

@@ -4,24 +4,22 @@ use std::sync::Arc;
 
 use crate::manager::deployment::DeploymentSearch;
 use crate::manager::fmt;
-use diesel::r2d2::ConnectionManager;
-use diesel::r2d2::PooledConnection;
-use diesel::PgConnection;
 use graph::components::store::DeploymentLocator;
 use graph::components::store::VersionStats;
 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::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>, PooledConnection<ConnectionManager<PgConnection>>), anyhow::Error> {
+) -> Result<(Arc<Site>, PgConnection), anyhow::Error> {
     let primary_pool = pools.get(&*PRIMARY_SHARD).unwrap();
     let locator = search.locate_unique(primary_pool).await?;
 

node/src/manager/commands/txn_speed.rs

@@ -1,8 +1,7 @@
-use diesel::PgConnection;
 use std::{collections::HashMap, thread::sleep, time::Duration};
 
 use graph::prelude::anyhow;
-use graph_store_postgres::ConnectionPool;
+use graph_store_postgres::{ConnectionPool, PgConnection};
 
 use crate::manager::catalog;
 

node/src/manager/deployment.rs

@@ -2,17 +2,21 @@ use std::collections::HashSet;
 use std::fmt;
 use std::str::FromStr;
 
-use diesel::{dsl::sql, prelude::*};
-use diesel::{sql_types::Text, PgConnection};
+use diesel::dsl::sql;
+use diesel::sql_types::Text;
+use diesel::{
+    ExpressionMethods, JoinOnDsl, NullableExpressionMethods, PgTextExpressionMethods, QueryDsl,
+    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::unused;
 use graph_store_postgres::ConnectionPool;
+use graph_store_postgres::{unused, PgConnection};
 
 lazy_static! {
     // `Qm...` optionally follow by `:$shard`

store/postgres/src/advisory_lock.rs

@@ -15,10 +15,11 @@
 //!           happens to it
 
 use diesel::sql_types::Bool;
-use diesel::{sql_query, PgConnection, RunQueryDsl};
+use diesel::{sql_query, RunQueryDsl};
 use graph::prelude::StoreError;
 
 use crate::command_support::catalog::Site;
+use crate::pool::PgConnection;
 use crate::primary::DeploymentId;
 
 /// A locking scope for a particular deployment. We use different scopes for

store/postgres/src/block_store.rs

@@ -6,10 +6,7 @@ use std::{
 
 use anyhow::anyhow;
 use async_trait::async_trait;
-use diesel::{
-    query_dsl::methods::FilterDsl as _, sql_query, ExpressionMethods as _, PgConnection,
-    RunQueryDsl,
-};
+use diesel::{query_dsl::methods::FilterDsl as _, sql_query, ExpressionMethods as _, RunQueryDsl};
 use diesel_async::scoped_futures::ScopedFutureExt;
 use graph::{
     blockchain::ChainIdentifier,
@@ -28,7 +25,7 @@ use graph::{prelude::StoreError, util::timed_cache::TimedCache};
 use crate::{
     chain_head_listener::ChainHeadUpdateSender,
     chain_store::{ChainStoreMetrics, Storage},
-    pool::ConnectionPool,
+    pool::{ConnectionPool, PgConnection},
     primary::Mirror as PrimaryMirror,
     ChainStore, NotificationSender, Shard, PRIMARY_SHARD,
 };
@@ -54,16 +51,15 @@ pub mod primary {
     use std::convert::TryFrom;
 
     use diesel::{
-        delete, insert_into, update, ExpressionMethods, OptionalExtension, PgConnection, QueryDsl,
-        RunQueryDsl,
+        delete, insert_into, update, ExpressionMethods, OptionalExtension, QueryDsl, RunQueryDsl,
     };
     use graph::{
         blockchain::{BlockHash, ChainIdentifier},
         internal_error,
         prelude::StoreError,
     };
 
-    use crate::chain_store::Storage;
+    use crate::{chain_store::Storage, pool::PgConnection};
     use crate::{ConnectionPool, Shard};
 
     table! {

store/postgres/src/catalog.rs

@@ -1,7 +1,7 @@
+use diesel::sql_query;
 use diesel::sql_types::{Bool, Integer};
 use diesel::{connection::SimpleConnection, prelude::RunQueryDsl, select};
 use diesel::{insert_into, OptionalExtension};
-use diesel::{pg::PgConnection, sql_query};
 use diesel::{
     sql_types::{Array, BigInt, Double, Nullable, Text},
     ExpressionMethods, QueryDsl,
@@ -22,6 +22,7 @@ use graph::{
     prelude::{lazy_static, StoreError, BLOCK_NUMBER_MAX},
 };
 
+use crate::pool::PgConnection;
 use crate::{
     block_range::BLOCK_RANGE_COLUMN,
     pool::ForeignServer,

store/postgres/src/chain_store.rs

@@ -1,10 +1,7 @@
 use anyhow::anyhow;
 use async_trait::async_trait;
-use diesel::pg::PgConnection;
-use diesel::prelude::*;
-use diesel::r2d2::{ConnectionManager, PooledConnection};
 use diesel::sql_types::Text;
-use diesel::{insert_into, update};
+use diesel::{insert_into, update, ExpressionMethods, OptionalExtension, QueryDsl, RunQueryDsl};
 use diesel_async::scoped_futures::ScopedFutureExt;
 
 use graph::components::store::ChainHeadStore;
@@ -37,7 +34,7 @@ use graph::prelude::{
 use graph::{ensure, internal_error};
 
 use self::recent_blocks_cache::RecentBlocksCache;
-use crate::pool::AsyncConnection;
+use crate::pool::{AsyncConnection, PgConnection};
 use crate::{
     block_store::ChainStatus, chain_head_listener::ChainHeadUpdateSender, pool::ConnectionPool,
 };
@@ -88,16 +85,18 @@ pub use data::Storage;
 /// Encapuslate access to the blocks table for a chain.
 mod data {
     use crate::diesel::dsl::IntervalDsl;
+    use diesel::dsl::sql;
     use diesel::sql_types::{Array, Binary, Bool, Nullable};
     use diesel::{connection::SimpleConnection, insert_into};
-    use diesel::{delete, prelude::*, sql_query};
+    use diesel::{
+        delete, sql_query, ExpressionMethods, JoinOnDsl, OptionalExtension, QueryDsl, RunQueryDsl,
+    };
     use diesel::{
         deserialize::FromSql,
         pg::Pg,
         serialize::{Output, ToSql},
         sql_types::Text,
     };
-    use diesel::{dsl::sql, pg::PgConnection};
     use diesel::{
         sql_types::{BigInt, Bytea, Integer, Jsonb},
         update,
@@ -119,6 +118,7 @@ mod data {
     use std::iter::FromIterator;
     use std::str::FromStr;
 
+    use crate::pool::PgConnection;
     use crate::transaction_receipt::RawTransactionReceipt;
 
     use super::JsonBlock;
@@ -1948,7 +1948,7 @@ impl ChainStore {
         matches!(self.status, ChainStatus::Ingestible)
     }
 
-    async fn get_conn(&self) -> Result<PooledConnection<ConnectionManager<PgConnection>>, Error> {
+    async fn get_conn(&self) -> Result<PgConnection, Error> {
         self.pool.get_sync().await.map_err(Error::from)
     }
 
@@ -3019,7 +3019,7 @@ impl EthereumCallCache for ChainStore {
         block: BlockPtr,
     ) -> Result<Option<call::Response>, Error> {
         let id = contract_call_id(req, &block);
-        let conn = &mut *self.get_conn().await?;
+        let conn = &mut self.get_conn().await?;
         let return_value = conn
             .transaction_async::<_, Error, _>(|conn| {
                 async {
@@ -3060,7 +3060,7 @@ impl EthereumCallCache for ChainStore {
             .collect();
         let id_refs: Vec<_> = ids.iter().map(|id| id.as_slice()).collect();
 
-        let conn = &mut *self.get_conn().await?;
+        let conn = &mut self.get_conn().await?;
         let rows = conn
             .transaction_async::<_, Error, _>(|conn| {
                 self.storage
@@ -3094,7 +3094,7 @@ impl EthereumCallCache for ChainStore {
     }
 
     async fn get_calls_in_block(&self, block: BlockPtr) -> Result<Vec<CachedEthereumCall>, Error> {
-        let conn = &mut *self.get_conn().await?;
+        let conn = &mut self.get_conn().await?;
         conn.transaction_async::<_, Error, _>(|conn| {
             self.storage.get_calls_in_block(conn, block).scope_boxed()
         })

store/postgres/src/copy.rs

@@ -23,12 +23,8 @@ use std::{
 };
 
 use diesel::{
-    connection::SimpleConnection as _,
-    dsl::sql,
-    insert_into,
-    r2d2::{ConnectionManager, PooledConnection},
-    select, sql_query, update, ExpressionMethods, OptionalExtension, PgConnection, QueryDsl,
-    RunQueryDsl,
+    connection::SimpleConnection as _, dsl::sql, insert_into, select, sql_query, update,
+    ExpressionMethods, OptionalExtension, QueryDsl, RunQueryDsl,
 };
 use diesel_async::scoped_futures::{ScopedBoxFuture, ScopedFutureExt};
 use graph::{
@@ -49,6 +45,7 @@ use itertools::Itertools;
 use crate::{
     advisory_lock, catalog, deployment,
     dynds::DataSourcesTable,
+    pool::PgConnection,
     primary::{DeploymentId, Primary, Site},
     relational::{index::IndexList, Layout, Table},
     relational_queries as rq,
@@ -679,12 +676,12 @@ impl From<Result<CopyTableWorker, StoreError>> for WorkerResult {
 /// This struct helps us with that. It wraps a connection and tracks whether
 /// the connection was used to acquire the copy lock
 struct LockTrackingConnection {
-    inner: PooledConnection<ConnectionManager<PgConnection>>,
+    inner: PgConnection,
     has_lock: bool,
 }
 
 impl LockTrackingConnection {
-    fn new(inner: PooledConnection<ConnectionManager<PgConnection>>) -> Self {
+    fn new(inner: PgConnection) -> Self {
         Self {
             inner,
             has_lock: false,

store/postgres/src/deployment.rs

@@ -1,8 +1,7 @@
 //! Utilities for dealing with deployment metadata. Any connection passed
 //! into these methods must be for the shard that holds the actual
 //! deployment data and metadata
-use crate::{advisory_lock, detail::GraphNodeVersion, primary::DeploymentId};
-use diesel::pg::PgConnection;
+use crate::{advisory_lock, detail::GraphNodeVersion, pool::PgConnection, primary::DeploymentId};
 use diesel::{
     connection::SimpleConnection,
     dsl::{count, delete, insert_into, now, select, sql, update},