Make all store/db interactions async

Cargo.toml

@@ -42,6 +42,7 @@ license = "MIT OR Apache-2.0"
 anyhow = "1.0"
 async-graphql = { version = "7.0.17", features = ["chrono"] }
 async-graphql-axum = "7.0.17"
+async-trait = "0.1.74"
 axum = "0.8.4"
 chrono = "0.4.42"
 bs58 = "0.5.1"
@@ -55,6 +56,7 @@ diesel = { version = "2.2.7", features = [
     "chrono",
     "i-implement-a-third-party-backend-and-opt-into-breaking-changes",
 ] }
+diesel-async = { version = "0.7.3", features = ["deadpool", "async-connection-wrapper", "tokio", "postgres"] }
 diesel-derive-enum = { version = "2.1.0", features = ["postgres"] }
 diesel-dynamic-schema = { version = "0.2.3", features = ["postgres"] }
 diesel_derives = "2.2.7"
@@ -90,7 +92,11 @@ strum = { version = "0.26", features = ["derive"] }
 syn = { version = "2.0.106", features = ["full"] }
 test-store = { path = "./store/test-store" }
 thiserror = "2.0.16"
+deadpool = { version = "0.12", features = ["rt_tokio_1", "managed"] }
 tokio = { version = "1.45.1", features = ["full"] }
+tokio-stream = { version = "0.1.15", features = ["sync"] }
+tokio-retry = "0.3.0"
+
 tonic = { version = "0.12.3", features = ["tls-roots", "gzip"] }
 tonic-build = { version = "0.12.3", features = ["prost"] }
 tower-http = { version = "0.6.6", features = ["cors"] }

chain/ethereum/Cargo.toml

@@ -4,6 +4,7 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
+async-trait = { workspace = true }
 envconfig = "0.11.0"
 jsonrpc-core = "18.0.0"
 graph = { path = "../../graph" }
@@ -15,6 +16,8 @@ tiny-keccak = "1.5.0"
 hex = "0.4.3"
 semver = "1.0.27"
 thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
 
 itertools = "0.14.0"
 

chain/ethereum/src/adapter.rs

@@ -1,4 +1,5 @@
 use anyhow::Error;
+use async_trait::async_trait;
 use ethabi::{Error as ABIError, ParamType, Token};
 use graph::blockchain::ChainIdentifier;
 use graph::components::subgraph::MappingError;

chain/ethereum/src/buffered_call_cache.rs

@@ -3,6 +3,7 @@ use std::{
     sync::{Arc, Mutex},
 };
 
+use async_trait::async_trait;
 use graph::{
     cheap_clone::CheapClone,
     components::store::EthereumCallCache,
@@ -47,8 +48,9 @@ impl BufferedCallCache {
     }
 }
 
+#[async_trait]
 impl EthereumCallCache for BufferedCallCache {
-    fn get_call(
+    async fn get_call(
         &self,
         call: &call::Request,
         block: BlockPtr,
@@ -59,7 +61,7 @@ impl EthereumCallCache for BufferedCallCache {
             return Ok(Some(value));
         }
 
-        let result = self.call_cache.get_call(&call, block)?;
+        let result = self.call_cache.get_call(&call, block).await?;
 
         let mut buffer = self.buffer.lock().unwrap();
         if let Some(call::Response {
@@ -73,7 +75,7 @@ impl EthereumCallCache for BufferedCallCache {
         Ok(result)
     }
 
-    fn get_calls(
+    async fn get_calls(
         &self,
         reqs: &[call::Request],
         block: BlockPtr,
@@ -90,7 +92,7 @@ impl EthereumCallCache for BufferedCallCache {
             }
         }
 
-        let (stored, calls) = self.call_cache.get_calls(&missing, block)?;
+        let (stored, calls) = self.call_cache.get_calls(&missing, block).await?;
 
         {
             let mut buffer = self.buffer.lock().unwrap();
@@ -103,15 +105,15 @@ impl EthereumCallCache for BufferedCallCache {
         Ok((resps, calls))
     }
 
-    fn get_calls_in_block(
+    async fn get_calls_in_block(
         &self,
         block: BlockPtr,
     ) -> Result<Vec<CachedEthereumCall>, graph::prelude::Error> {
-        self.call_cache.get_calls_in_block(block)
+        self.call_cache.get_calls_in_block(block).await
     }
 
-    fn set_call(
-        &self,
+    async fn set_call(
+        self: Arc<Self>,
         logger: &Logger,
         call: call::Request,
         block: BlockPtr,
@@ -130,15 +132,14 @@ impl EthereumCallCache for BufferedCallCache {
 
         let cache = self.call_cache.cheap_clone();
         let logger = logger.cheap_clone();
-        let _ = graph::spawn_blocking_allow_panic(move || {
-            cache
-                .set_call(&logger, call.cheap_clone(), block, return_value)
-                .map_err(|e| {
-                    error!(logger, "BufferedCallCache: call cache set error";
+        if let Err(e) = cache
+            .set_call(&logger, call.cheap_clone(), block, return_value)
+            .await
+        {
+            error!(logger, "BufferedCallCache: call cache set error";
                             "contract_address" => format!("{:?}", call.address),
                             "error" => e.to_string())
-                })
-        });
+        }
 
         Ok(())
     }

chain/ethereum/src/chain.rs

@@ -1,5 +1,6 @@
 use anyhow::{anyhow, bail, Result};
 use anyhow::{Context, Error};
+use async_trait::async_trait;
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::{FirehoseBlockIngestor, Transforms};
 use graph::blockchain::{
@@ -32,8 +33,8 @@ use graph::{
     components::store::DeploymentLocator,
     firehose,
     prelude::{
-        async_trait, o, serde_json as json, BlockNumber, ChainStore, EthereumBlockWithCalls,
-        Logger, LoggerFactory,
+        o, serde_json as json, BlockNumber, ChainStore, EthereumBlockWithCalls, Logger,
+        LoggerFactory,
     },
 };
 use prost::Message;
@@ -551,9 +552,11 @@ impl Blockchain for Chain {
         self.block_refetcher.get_block(self, logger, cursor).await
     }
 
-    fn runtime(&self) -> anyhow::Result<(Arc<dyn RuntimeAdapterTrait<Self>>, Self::DecoderHook)> {
+    async fn runtime(
+        &self,
+    ) -> anyhow::Result<(Arc<dyn RuntimeAdapterTrait<Self>>, Self::DecoderHook)> {
         let call_cache = Arc::new(BufferedCallCache::new(self.call_cache.cheap_clone()));
-        let chain_ident = self.chain_store.chain_identifier()?;
+        let chain_ident = self.chain_store.chain_identifier().await?;
 
         let builder = self.runtime_adapter_builder.build(
             self.eth_adapters.cheap_clone(),
@@ -1266,7 +1269,7 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
 #[cfg(test)]
 mod tests {
     use graph::blockchain::mock::MockChainStore;
-    use graph::{slog, tokio};
+    use graph::slog;
 
     use super::*;
     use std::sync::Arc;
@@ -1283,7 +1286,7 @@ mod tests {
         }
     }
 
-    #[tokio::test]
+    #[graph::test]
     async fn test_fetch_unique_blocks_single_block() {
         let logger = Logger::root(slog::Discard, o!());
         let mut chain_store = MockChainStore::default();
@@ -1302,7 +1305,7 @@ mod tests {
         assert!(missing.is_empty());
     }
 
-    #[tokio::test]
+    #[graph::test]
     async fn test_fetch_unique_blocks_duplicate_blocks() {
         let logger = Logger::root(slog::Discard, o!());
         let mut chain_store = MockChainStore::default();
@@ -1325,7 +1328,7 @@ mod tests {
         assert_eq!(missing[0], 1);
     }
 
-    #[tokio::test]
+    #[graph::test]
     async fn test_fetch_unique_blocks_missing_blocks() {
         let logger = Logger::root(slog::Discard, o!());
         let mut chain_store = MockChainStore::default();
@@ -1344,7 +1347,7 @@ mod tests {
         assert_eq!(missing, vec![2]);
     }
 
-    #[tokio::test]
+    #[graph::test]
     async fn test_fetch_unique_blocks_multiple_valid_blocks() {
         let logger = Logger::root(slog::Discard, o!());
         let mut chain_store = MockChainStore::default();
@@ -1366,7 +1369,7 @@ mod tests {
         assert!(missing.is_empty());
     }
 
-    #[tokio::test]
+    #[graph::test]
     async fn test_fetch_unique_blocks_mixed_scenario() {
         let logger = Logger::root(slog::Discard, o!());
         let mut chain_store = MockChainStore::default();

chain/ethereum/src/data_source.rs

@@ -1,5 +1,6 @@
 use anyhow::{anyhow, Error};
 use anyhow::{ensure, Context};
+use async_trait::async_trait;
 use graph::blockchain::{BlockPtr, TriggerWithHandler};
 use graph::components::link_resolver::LinkResolverContext;
 use graph::components::metrics::subgraph::SubgraphInstanceMetrics;
@@ -33,7 +34,6 @@ use tiny_keccak::{keccak256, Keccak};
 use graph::{
     blockchain::{self, Blockchain},
     prelude::{
-        async_trait,
         ethabi::{Address, Event, Function, LogParam, ParamType, RawLog},
         serde_json, warn,
         web3::types::{Log, Transaction, H256},

chain/ethereum/src/ethereum_adapter.rs

@@ -1,4 +1,8 @@
+use async_trait::async_trait;
 use futures03::{future::BoxFuture, stream::FuturesUnordered};
+use tokio::sync::RwLock;
+use tokio::time::timeout;
+
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockHash;
 use graph::blockchain::ChainIdentifier;
@@ -22,14 +26,11 @@ use graph::prelude::ethabi::Token;
 use graph::prelude::tokio::try_join;
 use graph::prelude::web3::types::U256;
 use graph::slog::o;
-use graph::tokio::sync::RwLock;
-use graph::tokio::time::timeout;
 use graph::{
     blockchain::{block_stream::BlockWithTriggers, BlockPtr, IngestorError},
     prelude::{
         anyhow::{self, anyhow, bail, ensure, Context},
-        async_trait, debug, error, ethabi, hex, info, retry, serde_json as json, tiny_keccak,
-        trace, warn,
+        debug, error, ethabi, hex, info, retry, serde_json as json, tiny_keccak, trace, warn,
         web3::{
             self,
             types::{
@@ -1653,6 +1654,7 @@ impl EthereumAdapterTrait for EthereumAdapter {
 
         let (mut resps, missing) = cache
             .get_calls(&reqs, block_ptr)
+            .await
             .map_err(|e| error!(logger, "call cache get error"; "error" => e.to_string()))
             .unwrap_or_else(|_| (Vec::new(), reqs));
 
@@ -1729,7 +1731,7 @@ impl EthereumAdapterTrait for EthereumAdapter {
             .iter()
             .map(|block| block as &dyn graph::blockchain::Block)
             .collect();
-        if let Err(e) = chain_store.upsert_light_blocks(block_refs.as_slice()) {
+        if let Err(e) = chain_store.upsert_light_blocks(block_refs.as_slice()).await {
             error!(logger, "Error writing to block cache {}", e);
         }
         blocks.extend(new_blocks);
@@ -2367,7 +2369,7 @@ async fn fetch_individual_receipts_with_retry(
     }
 
     // Use a stream to fetch receipts individually
-    let hash_stream = graph::tokio_stream::iter(hashes);
+    let hash_stream = tokio_stream::iter(hashes);
     let receipt_stream = hash_stream
         .map(move |tx_hash| {
             fetch_transaction_receipt_with_retry(
@@ -2379,7 +2381,7 @@ async fn fetch_individual_receipts_with_retry(
         })
         .buffered(ENV_VARS.block_ingestor_max_concurrent_json_rpc_calls);
 
-    graph::tokio_stream::StreamExt::collect::<Result<Vec<Arc<TransactionReceipt>>, IngestorError>>(
+    tokio_stream::StreamExt::collect::<Result<Vec<Arc<TransactionReceipt>>, IngestorError>>(
         receipt_stream,
     )
     .await
@@ -2672,7 +2674,6 @@ mod tests {
     };
     use graph::blockchain::BlockPtr;
     use graph::prelude::ethabi::ethereum_types::U64;
-    use graph::prelude::tokio::{self};
     use graph::prelude::web3::transports::test::TestTransport;
     use graph::prelude::web3::types::{Address, Block, Bytes, H256};
     use graph::prelude::web3::Web3;
@@ -2720,7 +2721,7 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[graph::test]
     async fn test_check_block_receipts_support() {
         let mut transport = TestTransport::default();
 

chain/ethereum/src/ingestor.rs

@@ -1,5 +1,6 @@
 use crate::{chain::BlockFinality, ENV_VARS};
 use crate::{EthereumAdapter, EthereumAdapterTrait as _};
+use async_trait::async_trait;
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockchainKind;
 use graph::components::network_provider::ChainName;
@@ -9,8 +10,8 @@ use graph::{
     blockchain::{BlockHash, BlockIngestor, BlockPtr, IngestorError},
     cheap_clone::CheapClone,
     prelude::{
-        async_trait, error, ethabi::ethereum_types::H256, info, tokio, trace, warn, ChainStore,
-        Error, EthereumBlockWithCalls, LogCode, Logger,
+        error, ethabi::ethereum_types::H256, info, tokio, trace, warn, ChainStore, Error,
+        EthereumBlockWithCalls, LogCode, Logger,
     },
 };
 use std::{sync::Arc, time::Duration};
@@ -43,8 +44,12 @@ impl PollingBlockIngestor {
         })
     }
 
-    fn cleanup_cached_blocks(&self) {
-        match self.chain_store.cleanup_cached_blocks(self.ancestor_count) {
+    async fn cleanup_cached_blocks(&self) {
+        match self
+            .chain_store
+            .cleanup_cached_blocks(self.ancestor_count)
+            .await
+        {
             Ok(Some((min_block, count))) => {
                 if count > 0 {
                     info!(
@@ -256,7 +261,7 @@ impl BlockIngestor for PollingBlockIngestor {
             }
 
             if ENV_VARS.cleanup_blocks {
-                self.cleanup_cached_blocks()
+                self.cleanup_cached_blocks().await
             }
 
             tokio::time::sleep(self.polling_interval).await;