store: Asyncify deployment::history_blocks

Author: lutter

store/postgres/src/deployment.rs

@@ -380,7 +380,10 @@ impl ManifestInfo {
 }
 
 // Return how many blocks of history this subgraph should keep
-pub fn history_blocks(conn: &mut PgConnection, site: &Site) -> Result<BlockNumber, StoreError> {
+pub async fn history_blocks(
+    conn: &mut PgConnection,
+    site: &Site,
+) -> Result<BlockNumber, StoreError> {
     use subgraph_manifest as sm;
     sm::table
         .select(sm::history_blocks)

store/postgres/src/relational.rs

@@ -37,12 +37,12 @@ use graph::data::graphql::TypeExt as _;
 use graph::data::query::Trace;
 use graph::data::value::Word;
 use graph::data_source::CausalityRegion;
-use graph::internal_error;
 use graph::prelude::{q, EntityQuery, StopwatchMetrics, ENV_VARS};
 use graph::schema::{
     EntityKey, EntityType, Field, FulltextConfig, FulltextDefinition, InputSchema,
 };
 use graph::slog::warn;
+use graph::{internal_error, tokio};
 use index::IndexList;
 use inflector::Inflector;
 use itertools::Itertools;
@@ -1087,13 +1087,13 @@ impl Layout {
     /// This is tied closely to how the `LayoutCache` works and called from
     /// it right after creating a `Layout`, and periodically to update the
     /// `Layout` in case changes were made
-    fn refresh(
+    async fn refresh(
         self: Arc<Self>,
         conn: &mut PgConnection,
         site: Arc<Site>,
     ) -> Result<Arc<Self>, StoreError> {
         let account_like = crate::catalog::account_like(conn, &self.site)?;
-        let history_blocks = deployment::history_blocks(conn, &self.site)?;
+        let history_blocks = deployment::history_blocks(conn, &self.site).await?;
 
         let is_account_like = { |table: &Table| account_like.contains(table.name.as_str()) };
 
@@ -1750,7 +1750,7 @@ pub struct LayoutCache {
     ttl: Duration,
     /// Use this so that we only refresh one layout at any given time to
     /// avoid refreshing the same layout multiple times
-    refresh: Mutex<()>,
+    refresh: tokio::sync::Mutex<()>,
     last_sweep: Mutex<Instant>,
 }
 
@@ -1759,7 +1759,7 @@ impl LayoutCache {
         Self {
             entries: Mutex::new(HashMap::new()),
             ttl,
-            refresh: Mutex::new(()),
+            refresh: tokio::sync::Mutex::new(()),
             last_sweep: Mutex::new(Instant::now()),
         }
     }
@@ -1770,7 +1770,7 @@ impl LayoutCache {
             deployment::entities_with_causality_region(conn, site.id, &subgraph_schema)?;
         let catalog = Catalog::load(conn, site.clone(), use_bytea_prefix, has_causality_region)?;
         let layout = Arc::new(Layout::new(site.clone(), &subgraph_schema, catalog)?);
-        layout.refresh(conn, site)
+        layout.refresh(conn, site).await
     }
 
     fn cache(&self, layout: Arc<Layout>) {
@@ -1822,7 +1822,7 @@ impl LayoutCache {
                     if refresh.is_err() {
                         value
                     } else {
-                        self.refresh(logger, conn, site, value)
+                        self.refresh(logger, conn, site, value).await
                     }
                 }
             }
@@ -1836,14 +1836,14 @@ impl LayoutCache {
         Ok(layout)
     }
 
-    fn refresh(
+    async fn refresh(
         &self,
         logger: &Logger,
         conn: &mut PgConnection,
         site: Arc<Site>,
         value: Arc<Layout>,
     ) -> Arc<Layout> {
-        match value.cheap_clone().refresh(conn, site) {
+        match value.cheap_clone().refresh(conn, site).await {
             Err(e) => {
                 warn!(
                     logger,