fix: Adding lock to init relayer instances (#622)

* fix: Adding lock to init relayer instances

* fix: Fixing test

* test: Adding test cases

* test: Adding coverage

* test: Adding coverage

* test: Adding test cases

* feat: Global lock for init and config file processing

* fix: Improvements

* fix: Suggestions

* fix: Suggestions

* chore: Use distributed mode flag

* fix: Improvements

* fix: Fixing redis config validation for dist mode

* feat: Added logic to recovery after timeout

* test: Test coverage

* fix: Config processing path locks

* fix: Change wording to avoid codeQL false positive

---------

Co-authored-by: Zeljko <zeljko89markovic@gmail.com>
Co-authored-by: tirumerla <tirumerla@gmail.com>

Author: 3 people

src/bootstrap/config_processor.rs

@@ -330,54 +330,54 @@ impl ApiKeyRepositoryTrait for RedisApiKeyRepository {
             .await?;
         let plugin_list_key = self.api_key_list_key();
 
-        debug!("Checking if plugin entries exist");
+        debug!("Checking if API key entries exist");
 
         let exists: bool = conn
             .exists(&plugin_list_key)
             .await
             .map_err(|e| self.map_redis_error(e, "has_entries_check"))?;
 
-        debug!("Plugin entries exist: {}", exists);
+        debug!("API key entries exist: {}", exists);
         Ok(exists)
     }
 
     async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
         let mut conn = self
             .get_connection(self.connections.primary(), "drop_all_entries")
             .await?;
-        let plugin_list_key = self.api_key_list_key();
+        let api_key_list_key = self.api_key_list_key();
 
-        debug!("Dropping all plugin entries");
+        debug!("Dropping all API key entries");
 
-        // Get all plugin IDs first
-        let plugin_ids: Vec<String> = conn
-            .smembers(&plugin_list_key)
+        // Get all API key IDs first
+        let api_key_ids: Vec<String> = conn
+            .smembers(&api_key_list_key)
             .await
             .map_err(|e| self.map_redis_error(e, "drop_all_entries_get_ids"))?;
 
-        if plugin_ids.is_empty() {
-            debug!("No plugin entries to drop");
+        if api_key_ids.is_empty() {
+            debug!("No API key entries to drop");
             return Ok(());
         }
 
         // Use pipeline for atomic operations
         let mut pipe = redis::pipe();
         pipe.atomic();
 
-        // Delete all individual plugin entries
-        for plugin_id in &plugin_ids {
-            let plugin_key = self.api_key_key(plugin_id);
-            pipe.del(&plugin_key);
+        // Delete all individual API key entries
+        for api_key_id in &api_key_ids {
+            let api_key_key = self.api_key_key(api_key_id);
+            pipe.del(&api_key_key);
         }
 
-        // Delete the plugin list key
-        pipe.del(&plugin_list_key);
+        // Delete the API key list key
+        pipe.del(&api_key_list_key);
 
         pipe.exec_async(&mut conn)
             .await
             .map_err(|e| self.map_redis_error(e, "drop_all_entries_pipeline"))?;
 
-        debug!("Dropped {} plugin entries", plugin_ids.len());
+        debug!("Dropped {} API key entries", api_key_ids.len());
         Ok(())
     }
 }

src/bootstrap/initialize_relayers.rs

@@ -34,6 +34,7 @@ use crate::{
     utils::RedisConnections,
 };
 use async_trait::async_trait;
+use deadpool_redis::Pool;
 use std::sync::Arc;
 
 #[async_trait]
@@ -67,6 +68,19 @@ pub trait RelayerRepository: Repository<RelayerRepoModel, String> + Send + Sync
     /// Returns true if this repository uses persistent storage (e.g., Redis).
     /// Returns false for in-memory storage.
     fn is_persistent_storage(&self) -> bool;
+
+    /// Returns connection info for distributed operations.
+    ///
+    /// This method provides access to the underlying connection and key prefix
+    /// when using persistent storage. This is useful for distributed locking and
+    /// other coordination operations that need direct storage access.
+    ///
+    /// # Returns
+    /// * `Some((pool, prefix))` - If using persistent storage (e.g., Redis)
+    /// * `None` - If using in-memory storage (default)
+    fn connection_info(&self) -> Option<(Arc<Pool>, String)> {
+        None
+    }
 }
 
 #[cfg(test)]
@@ -96,6 +110,7 @@ mockall::mock! {
         async fn disable_relayer(&self, relayer_id: String, reason: DisabledReason) -> Result<RelayerRepoModel, RepositoryError>;
         async fn update_policy(&self, id: String, policy: RelayerNetworkPolicy) -> Result<RelayerRepoModel, RepositoryError>;
         fn is_persistent_storage(&self) -> bool;
+        fn connection_info(&self) -> Option<(Arc<Pool>, String)>;
     }
 }
 
@@ -120,6 +135,24 @@ impl RelayerRepositoryStorage {
             key_prefix,
         )?))
     }
+
+    /// Returns connection info for distributed operations.
+    ///
+    /// This method provides access to the underlying Redis connection and key prefix
+    /// when using Redis-backed storage. This is useful for distributed locking and
+    /// other coordination operations that need direct Redis access.
+    ///
+    /// # Returns
+    /// * `Some((pool, prefix))` - If using persistent storage (e.g., Redis)
+    /// * `None` - If using in-memory storage
+    pub fn connection_info(&self) -> Option<(Arc<Pool>, String)> {
+        match self {
+            RelayerRepositoryStorage::InMemory(_) => None,
+            RelayerRepositoryStorage::Redis(repo) => {
+                Some((repo.connections.primary().clone(), repo.key_prefix.clone()))
+            }
+        }
+    }
 }
 
 impl Default for RelayerRepositoryStorage {
@@ -285,6 +318,15 @@ impl RelayerRepository for RelayerRepositoryStorage {
             RelayerRepositoryStorage::Redis(_) => true,
         }
     }
+
+    fn connection_info(&self) -> Option<(Arc<Pool>, String)> {
+        match self {
+            RelayerRepositoryStorage::InMemory(_) => None,
+            RelayerRepositoryStorage::Redis(repo) => {
+                Some((repo.connections.primary().clone(), repo.key_prefix.clone()))
+            }
+        }
+    }
 }
 
 #[cfg(test)]
@@ -499,4 +541,38 @@ mod tests {
         repo.drop_all_entries().await.unwrap();
         assert!(!repo.has_entries().await.unwrap());
     }
+
+    #[tokio::test]
+    async fn test_connection_info_returns_none_for_in_memory() {
+        let storage = RelayerRepositoryStorage::new_in_memory();
+
+        // In-memory storage should return None for connection_info
+        assert!(storage.connection_info().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_is_persistent_storage_returns_false_for_in_memory() {
+        let storage = RelayerRepositoryStorage::new_in_memory();
+
+        // In-memory storage should return false for is_persistent_storage
+        assert!(!storage.is_persistent_storage());
+    }
+
+    #[tokio::test]
+    async fn test_trait_connection_info_returns_none_for_in_memory() {
+        let storage = RelayerRepositoryStorage::new_in_memory();
+
+        // Test the RelayerRepository trait's connection_info method
+        let trait_ref: &dyn RelayerRepository = &storage;
+        assert!(trait_ref.connection_info().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_struct_connection_info_returns_none_for_in_memory() {
+        let storage = RelayerRepositoryStorage::new_in_memory();
+
+        // Test the struct's own connection_info method
+        let result: Option<(Arc<Pool>, String)> = storage.connection_info();
+        assert!(result.is_none());
+    }
 }

src/repositories/api_key/api_key_redis.rs

@@ -166,6 +166,8 @@ impl RelayerRepository for InMemoryRelayerRepository {
     fn is_persistent_storage(&self) -> bool {
         false
     }
+
+    // Uses the trait default implementation which returns None
 }
 
 #[async_trait]

src/repositories/relayer/mod.rs

@@ -560,6 +560,10 @@ impl RelayerRepository for RedisRelayerRepository {
     fn is_persistent_storage(&self) -> bool {
         true
     }
+
+    fn connection_info(&self) -> Option<(Arc<deadpool_redis::Pool>, String)> {
+        Some((self.connections.primary().clone(), self.key_prefix.clone()))
+    }
 }
 
 #[cfg(test)]

src/repositories/relayer/relayer_in_memory.rs

@@ -8,6 +8,9 @@ pub use key::*;
 mod auth;
 pub use auth::*;
 
+mod polling;
+pub use polling::*;
+
 mod time;
 pub use time::*;