refactor(torii-core): add provider health check (#2729)

* [ISSUES#]2721 health_check_provider. * [ISSUES#]2721 health_check_provider. * [ISSUES#]2721 health_check_provider. * fix: change info to trace and refacto structured logging --------- Co-authored-by: glihm <[email protected]>
dojoengine · Nov 29, 2024 · 1aab149 · 1aab149
1 parent 08a52c7
commit 1aab149
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 2 deletions.
diff --git a/crates/torii/core/src/engine.rs b/crates/torii/core/src/engine.rs
@@ -45,6 +45,7 @@ use crate::processors::{
 };
 use crate::sql::{Cursors, Sql};
 use crate::types::{Contract, ContractType};
+use crate::utils::health_check_provider;
 
 type EventProcessorMap<P> = HashMap<Felt, Vec<Box<dyn EventProcessor<P>>>>;
 
@@ -241,6 +242,11 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
     }
 
     pub async fn start(&mut self) -> Result<()> {
+        if let Err(e) = health_check_provider(self.provider.clone()).await {
+            error!(target: LOG_TARGET,"Provider health check failed during engine start");
+            return Err(e);
+        }
+
         let mut backoff_delay = Duration::from_secs(1);
         let max_backoff_delay = Duration::from_secs(60);
 

diff --git a/crates/torii/core/src/utils.rs b/crates/torii/core/src/utils.rs
@@ -4,8 +4,10 @@ use anyhow::Result;
 use chrono::{DateTime, Utc};
 use futures_util::TryStreamExt;
 use ipfs_api_backend_hyper::{IpfsApi, IpfsClient, TryFromUri};
+use starknet::core::types::{BlockId, BlockTag};
+use starknet::providers::Provider;
 use tokio_util::bytes::Bytes;
-use tracing::info;
+use tracing::{info, trace};
 
 use crate::constants::{
     IPFS_CLIENT_MAX_RETRY, IPFS_CLIENT_PASSWORD, IPFS_CLIENT_URL, IPFS_CLIENT_USERNAME,
@@ -20,7 +22,6 @@ pub fn must_utc_datetime_from_timestamp(timestamp: u64) -> DateTime<Utc> {
 pub fn utc_dt_string_from_timestamp(timestamp: u64) -> String {
     must_utc_datetime_from_timestamp(timestamp).to_rfc3339()
 }
-
 pub async fn fetch_content_from_ipfs(cid: &str, mut retries: u8) -> Result<Bytes> {
     let client = IpfsClient::from_str(IPFS_CLIENT_URL)?
         .with_credentials(IPFS_CLIENT_USERNAME, IPFS_CLIENT_PASSWORD);
@@ -46,6 +47,26 @@ pub async fn fetch_content_from_ipfs(cid: &str, mut retries: u8) -> Result<Bytes
         IPFS_CLIENT_MAX_RETRY, cid
     )))
 }
+
+pub async fn health_check_provider<P: Provider + Sync + std::fmt::Debug + 'static>(
+    provider: P,
+) -> Result<(), anyhow::Error> {
+    match provider.get_block_with_tx_hashes(BlockId::Tag(BlockTag::Latest)).await {
+        Ok(block) => {
+            trace!(
+                latest_block = ?block,
+                "Provider health check."
+            );
+            Ok(())
+        }
+        Err(_) => {
+            let error_info =
+                format!("Unhealthy provider {:?}, please check your configuration.", provider);
+            Err(anyhow::anyhow!(error_info))
+        }
+    }
+}
+
 // tests
 #[cfg(test)]
 mod tests {