fix(omni-relayer): added timeout to redis read related queries (#530)

* fix: added timeout to redis read related queries

* chore: bumped relayer version

* refactor: use `set_response_timeout` during connection manager construction

Author: frolvanya

omni-relayer/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "omni-relayer"
-version = "0.5.11"
+version = "0.5.12"
 edition = "2024"
 resolver = "2"
 repository = "https://github.com/Near-One/omni-bridge"

omni-relayer/Cargo.toml

@@ -3,6 +3,7 @@ url = "redis://127.0.0.1/"
 sleep_time_after_events_process_secs = 1
 query_retry_attempts = 10
 query_retry_sleep_secs = 1
+query_timeout_secs = 5
 fee_retry_base_secs = 1.2
 fee_retry_max_sleep_secs = 3600 # 60 * 60
 keep_transfers_for_secs = 604800 # 60 * 60 * 24 * 7

omni-relayer/example-devnet-config.toml

@@ -3,6 +3,7 @@ url = "redis://127.0.0.1/"
 sleep_time_after_events_process_secs = 1
 query_retry_attempts = 10
 query_retry_sleep_secs = 1
+query_timeout_secs = 5
 fee_retry_base_secs = 1.2
 fee_retry_max_sleep_secs = 3600 # 60 * 60
 keep_transfers_for_secs = 604800 # 60 * 60 * 24 * 7

omni-relayer/example-mainnet-config.toml

@@ -3,6 +3,7 @@ url = "redis://127.0.0.1/"
 sleep_time_after_events_process_secs = 1
 query_retry_attempts = 10
 query_retry_sleep_secs = 1
+query_timeout_secs = 5
 fee_retry_base_secs = 1.2
 fee_retry_max_sleep_secs = 3600 # 60 * 60
 keep_transfers_for_secs = 604800 # 60 * 60 * 24 * 7

omni-relayer/example-testnet-config.toml

@@ -176,6 +176,7 @@ pub struct Redis {
     pub sleep_time_after_events_process_secs: u64,
     pub query_retry_attempts: u64,
     pub query_retry_sleep_secs: u64,
+    pub query_timeout_secs: u64,
     pub fee_retry_base_secs: Decimal,
     pub fee_retry_max_sleep_secs: i64,
     pub keep_transfers_for_secs: i64,

omni-relayer/src/config.rs

@@ -94,6 +94,20 @@ fn init_logging(network: Network) -> Result<()> {
     Ok(())
 }
 
+async fn build_redis_connection_manager(
+    config: &config::Config,
+) -> Result<redis::aio::ConnectionManager> {
+    let redis_client = redis::Client::open(config.redis.url.clone())?;
+    let redis_connection_manager = redis::aio::ConnectionManager::new_with_config(
+        redis_client,
+        redis::aio::ConnectionManagerConfig::new().set_response_timeout(
+            tokio::time::Duration::from_secs(config.redis.query_timeout_secs),
+        ),
+    )
+    .await?;
+    Ok(redis_connection_manager)
+}
+
 #[tokio::main]
 async fn main() -> Result<()> {
     dotenv::dotenv().ok();
@@ -111,8 +125,9 @@ async fn main() -> Result<()> {
 
     init_logging(config.near.network).context("Failed to initialize logging")?;
 
-    let redis_client = redis::Client::open(config.redis.url.clone())?;
-    let redis_connection_manager = redis::aio::ConnectionManager::new(redis_client.clone()).await?;
+    let redis_connection_manager = build_redis_connection_manager(&config)
+        .await
+        .context("Failed to create Redis connection manager")?;
     let jsonrpc_client = near_jsonrpc_client::JsonRpcClient::connect(config.near.rpc_url.clone());
 
     let near_omni_signer = startup::near::get_signer(&config, config::NearSignerType::Omni)?;

omni-relayer/src/main.rs

@@ -115,23 +115,39 @@ pub async fn get_events(
     key: String,
 ) -> Option<Vec<(String, String)>> {
     for _ in 0..config.redis.query_retry_attempts {
-        if let Ok(mut iter) = redis_connection_manager
+        let mut iter = match redis_connection_manager
             .hscan::<String, (String, String)>(key.clone())
             .await
         {
-            let mut events = Vec::new();
-
-            while let Some(event) = iter.next_item().await {
-                events.push(event);
+            Ok(iter) => iter,
+            Err(err) => {
+                warn!("Redis hscan failed: {err:?}");
+                tokio::time::sleep(tokio::time::Duration::from_secs(
+                    config.redis.query_retry_sleep_secs,
+                ))
+                .await;
+                continue;
+            }
+        };
+
+        let mut events = Vec::new();
+        loop {
+            match tokio::time::timeout(
+                tokio::time::Duration::from_secs(config.redis.query_timeout_secs),
+                iter.next_item(),
+            )
+            .await
+            {
+                Ok(Some(event)) => events.push(event),
+                Ok(None) => break,
+                Err(_) => {
+                    warn!("Redis hscan iteration timed out");
+                    break;
+                }
             }
-
-            return Some(events);
         }
 
-        tokio::time::sleep(tokio::time::Duration::from_secs(
-            config.redis.query_retry_sleep_secs,
-        ))
-        .await;
+        return Some(events);
     }
 
     warn!("Failed to get events from redis db");