store: Get AsyncPgConnection from pool in DeploymentStore

Author: lutter

store/postgres/src/deployment_store.rs

@@ -47,7 +47,6 @@ use crate::block_range::{BLOCK_COLUMN, BLOCK_RANGE_COLUMN};
 use crate::deployment::{self, OnSync};
 use crate::detail::ErrorDetail;
 use crate::dynds::DataSourcesTable;
-use crate::pool::PgConnection;
 use crate::primary::{DeploymentId, Primary};
 use crate::relational::index::{CreateIndex, IndexList, Method};
 use crate::relational::{self, Layout, LayoutCache, SqlName, Table};
@@ -353,11 +352,6 @@ impl DeploymentStore {
         Ok(count)
     }
 
-    /// Deprecated. Use `with_conn` instead.
-    async fn get_conn(&self) -> Result<PgConnection, StoreError> {
-        self.pool.get_sync().await
-    }
-
     /// Panics if `idx` is not a valid index for a read only pool.
     async fn read_only_conn(&self, idx: usize) -> Result<AsyncPgConnection, Error> {
         self.read_only_pools[idx].get().await.map_err(Error::from)
@@ -981,7 +975,7 @@ impl DeploymentStore {
         if ids_for_type.is_empty() {
             return Ok(BTreeMap::new());
         }
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         let layout = self.layout(&mut conn, site).await?;
 
         layout.find_many(&mut conn, ids_for_type, block).await
@@ -994,7 +988,7 @@ impl DeploymentStore {
         causality_region: CausalityRegion,
         block_range: Range<BlockNumber>,
     ) -> Result<BTreeMap<BlockNumber, Vec<EntitySourceOperation>>, StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         let layout = self.layout(&mut conn, site).await?;
         layout
             .find_range(&mut conn, entity_types, causality_region, block_range)
@@ -1008,7 +1002,7 @@ impl DeploymentStore {
         block: BlockNumber,
         excluded_keys: &Vec<EntityKey>,
     ) -> Result<BTreeMap<EntityKey, Entity>, StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         let layout = self.layout(&mut conn, site).await?;
         layout
             .find_derived(&mut conn, derived_query, block, excluded_keys)
@@ -1020,7 +1014,7 @@ impl DeploymentStore {
         site: Arc<Site>,
         block: BlockNumber,
     ) -> Result<Vec<EntityOperation>, StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         let layout = self.layout(&mut conn, site).await?;
         let changes = layout.find_changes(&mut conn, block).await?;
 
@@ -1034,7 +1028,7 @@ impl DeploymentStore {
         site: Arc<Site>,
         query: EntityQuery,
     ) -> Result<Vec<Entity>, QueryExecutionError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         self.execute_query(&mut conn, site, query)
             .await
             .map(|(entities, _)| entities)
@@ -1051,7 +1045,7 @@ impl DeploymentStore {
     ) -> Result<(), StoreError> {
         let mut conn = {
             let _section = stopwatch.start_section("transact_blocks_get_conn");
-            self.get_conn().await?
+            self.pool.get().await?
         };
 
         let (layout, earliest_block) = deployment::with_lock(&mut conn, &site, async |conn| {
@@ -1285,7 +1279,7 @@ impl DeploymentStore {
         site: Arc<Site>,
         block_ptr_to: BlockPtr,
     ) -> Result<(), StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
 
         let block_ptr_from = Self::block_ptr_with_conn(&mut conn, site.cheap_clone()).await?;
 
@@ -1316,7 +1310,7 @@ impl DeploymentStore {
         site: Arc<Site>,
         block_ptr_to: BlockPtr,
     ) -> Result<(), StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
 
         let block_ptr_from = Self::block_ptr_with_conn(&mut conn, site.cheap_clone()).await?;
 
@@ -1348,7 +1342,7 @@ impl DeploymentStore {
         block_ptr_to: BlockPtr,
         firehose_cursor: &FirehoseCursor,
     ) -> Result<(), StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         // Unwrap: If we are reverting then the block ptr is not `None`.
         let deployment_head = Self::block_ptr_with_conn(&mut conn, site.cheap_clone())
             .await?
@@ -1499,7 +1493,7 @@ impl DeploymentStore {
                 return Err(StoreError::Canceled);
             }
 
-            let mut conn = self.get_conn().await?;
+            let mut conn = self.pool.get().await?;
             conn.transaction::<(), StoreError, _>(|conn| {
                 async {
                     // Copy shared dynamic data sources and adjust their ID; if
@@ -1573,7 +1567,7 @@ impl DeploymentStore {
             .await?;
         }
 
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         if ENV_VARS.postpone_attribute_index_creation {
             // check if all indexes are valid and recreate them if they aren't
             self.load_indexes(site.clone())
@@ -1608,7 +1602,7 @@ impl DeploymentStore {
         current_ptr: &BlockPtr,
         parent_ptr: &BlockPtr,
     ) -> Result<UnfailOutcome, StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         let deployment_id = &site.deployment;
 
         conn.transaction(|conn| {
@@ -1705,7 +1699,7 @@ impl DeploymentStore {
         site: Arc<Site>,
         current_ptr: &BlockPtr,
     ) -> Result<UnfailOutcome, StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         let deployment_id = &site.deployment;
 
         conn.transaction(|conn| async {
@@ -1768,7 +1762,7 @@ impl DeploymentStore {
 
     #[cfg(debug_assertions)]
     pub async fn error_count(&self, id: &DeploymentHash) -> Result<usize, StoreError> {
-        let mut conn = self.get_conn().await?;
+        let mut conn = self.pool.get().await?;
         deployment::error_count(&mut conn, id).await
     }