From 77ef5d9895a6adeec7051bfe75744e5bf4ad7873 Mon Sep 17 00:00:00 2001 From: Collins Date: Fri, 13 Mar 2026 17:49:14 +0300 Subject: [PATCH] Revert "fix: Adding lock to init relayer instances (#622)" This reverts commit 51df3c5d43ada37f6c73dc0f16a557f9086aba32. --- src/bootstrap/config_processor.rs | 424 +------- src/bootstrap/initialize_relayers.rs | 922 +---------------- src/repositories/api_key/api_key_redis.rs | 32 +- src/repositories/relayer/mod.rs | 76 -- src/repositories/relayer/relayer_in_memory.rs | 2 - src/repositories/relayer/relayer_redis.rs | 4 - src/utils/mod.rs | 3 - src/utils/polling.rs | 192 ---- src/utils/redis.rs | 953 ------------------ 9 files changed, 46 insertions(+), 2562 deletions(-) delete mode 100644 src/utils/polling.rs diff --git a/src/bootstrap/config_processor.rs b/src/bootstrap/config_processor.rs index 4237623ef..5b404690c 100644 --- a/src/bootstrap/config_processor.rs +++ b/src/bootstrap/config_processor.rs @@ -1,20 +1,6 @@ //! This module provides functionality for processing configuration files and populating //! repositories. -//! -//! ## Distributed Locking for Config Processing -//! -//! When multiple instances of the relayer service start simultaneously with Redis storage -//! and `DISTRIBUTED_MODE` is enabled, this module uses distributed locking to coordinate -//! config processing and prevent race conditions: -//! -//! - **Global lock**: A single lock is used for the entire config processing, -//! ensuring only one instance processes the config at a time. -//! - **Post-lock population check**: After acquiring the lock, checks if Redis is already -//! populated (by another instance that held the lock first), and skips if so. -//! - **Single-instance mode**: When `DISTRIBUTED_MODE` is disabled (default) or using -//! in-memory storage, locking is skipped since coordination is not needed. use std::sync::Arc; -use std::time::Duration; use crate::{ config::{Config, RepositoryStorageType, ServerConfig}, @@ -29,19 +15,10 @@ use crate::{ Repository, TransactionCounterTrait, TransactionRepository, }, services::signer::{Signer as SignerService, SignerFactory}, - utils::{ - is_config_processing_completed, poll_until, set_config_processing_completed, - set_config_processing_in_progress, DistributedLock, BOOTSTRAP_LOCK_TTL_SECS, - LOCK_POLL_INTERVAL_MS, LOCK_WAIT_MAX_SECS, - }, }; use color_eyre::{eyre::WrapErr, Report, Result}; -use deadpool_redis::Pool; use futures::future::try_join_all; -use tracing::{info, warn}; - -/// Lock name for config processing lock. -const CONFIG_PROCESSING_LOCK_NAME: &str = "config_processing"; +use tracing::info; async fn process_api_key( server_config: &ServerConfig, @@ -345,10 +322,6 @@ where return Ok(true); } - if app_state.api_key_repository.has_entries().await? { - return Ok(true); - } - Ok(false) } @@ -361,10 +334,6 @@ where /// 4. Process networks /// 5. Process relayers /// 6. Process API key -/// -/// When using Redis storage with `DISTRIBUTED_MODE` enabled, this function uses distributed -/// locking to prevent race conditions when multiple instances start simultaneously -/// (especially with `RESET_STORAGE_ON_START=true`). pub async fn process_config_file( config_file: Config, server_config: Arc, @@ -381,300 +350,18 @@ where PR: PluginRepositoryTrait + Send + Sync + 'static, AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, { - match server_config.repository_storage_type { - RepositoryStorageType::InMemory => { - // In-memory mode: no locking needed, process directly - execute_config_processing(&config_file, &server_config, app_state).await - } + let should_process_config_file = match server_config.repository_storage_type { + RepositoryStorageType::InMemory => true, RepositoryStorageType::Redis => { - // Check if distributed locking is needed - let use_lock = ServerConfig::get_distributed_mode(); - let connection_info = app_state.relayer_repository.connection_info(); - - match (use_lock, connection_info) { - (true, Some((conn, prefix))) => { - // Distributed mode: use locking to coordinate across instances - coordinate_config_with_lock( - &config_file, - &server_config, - app_state, - &conn, - &prefix, - ) - .await - } - _ => { - // Single-instance mode or no connection info: skip locking - let should_process = server_config.reset_storage_on_start - || !is_redis_populated(app_state).await?; - - if should_process { - execute_config_processing(&config_file, &server_config, app_state).await - } else { - info!("Skipping config file processing - Redis already populated"); - Ok(()) - } - } - } - } - } -} - -/// Process config file with distributed locking for Redis storage. -/// -/// Flow: -/// 1. Try to acquire global lock for config processing -/// 2. If lock acquired: check for an explicit completion marker, process if needed -/// 3. If lock held: wait for the completion marker to appear -/// 4. If wait times out: recheck state and attempt takeover after lock expiry -async fn coordinate_config_with_lock( - config_file: &Config, - server_config: &ServerConfig, - app_state: &ThinDataAppState, - conn: &Arc, - prefix: &str, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - let lock_key = format!("{prefix}:lock:{CONFIG_PROCESSING_LOCK_NAME}"); - let lock = DistributedLock::new( - conn.clone(), - &lock_key, - Duration::from_secs(BOOTSTRAP_LOCK_TTL_SECS), - ); - - match lock.try_acquire().await { - Ok(Some(guard)) => { - // We got the lock - check if we need to process - info!("Acquired config processing lock"); - - let result = - process_if_needed_after_lock(config_file, server_config, app_state, conn, prefix) - .await; - - drop(guard); // Release lock - result - } - Ok(None) => { - // Lock held by another instance - wait for it to complete - info!("Another instance is processing config, waiting for completion"); - let completed = wait_for_config_processing_complete(conn, prefix).await?; - - if completed { - return Ok(()); - } - - warn!("Timeout waiting for config processing, rechecking state"); - - if is_config_processing_completed(conn, prefix) - .await - .unwrap_or(false) - { - info!("Config processing completed during timeout window"); - return Ok(()); - } - - recover_config_processing_after_timeout( - config_file, - server_config, - app_state, - conn, - prefix, - ) - .await - } - Err(e) => { - // Lock error - graceful degradation, proceed without lock - warn!( - error = %e, - "Failed to acquire config processing lock, proceeding without coordination" - ); - execute_config_processing(config_file, server_config, app_state).await - } - } -} - -/// Process config after successfully acquiring the lock. -/// -/// Checks if config processing was already completed and only processes if needed. -async fn process_if_needed_after_lock( - config_file: &Config, - server_config: &ServerConfig, - app_state: &ThinDataAppState, - conn: &Arc, - prefix: &str, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - let already_completed = is_config_processing_completed(conn, prefix).await?; - - if server_config.reset_storage_on_start { - // With reset flag: always reset and process (we have the lock) - execute_config_processing_with_marker(config_file, server_config, app_state, conn, prefix) - .await - } else if already_completed { - // No reset flag and already completed: skip - info!("Config processing already completed, skipping config file processing"); - Ok(()) - } else { - // No reset flag and not completed: process - execute_config_processing_with_marker(config_file, server_config, app_state, conn, prefix) - .await - } -} - -/// Waits for another instance to complete config processing. -/// -/// Polls periodically until the explicit completion marker is set or timeout is reached. -async fn wait_for_config_processing_complete(conn: &Arc, prefix: &str) -> Result { - let max_wait = Duration::from_secs(LOCK_WAIT_MAX_SECS); - let poll_interval = Duration::from_millis(LOCK_POLL_INTERVAL_MS); - - let conn = conn.clone(); - let prefix = prefix.to_string(); - - let completed = poll_until( - || is_config_processing_completed(&conn, &prefix), - max_wait, - poll_interval, - "config processing", - ) - .await?; - - Ok(completed) -} - -/// Attempts to recover config processing after a wait timeout. -/// -/// This is the config-processing analogue to relayer initialization recovery: -/// if the original lock holder died after setting the in-progress marker but -/// before completion, a waiting instance should take over once the lock TTL -/// expires instead of failing the whole rollout. -async fn recover_config_processing_after_timeout( - config_file: &Config, - server_config: &ServerConfig, - app_state: &ThinDataAppState, - conn: &Arc, - prefix: &str, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - let lock_key = format!("{prefix}:lock:{CONFIG_PROCESSING_LOCK_NAME}"); - let recovery_lock = DistributedLock::new( - conn.clone(), - &lock_key, - Duration::from_secs(BOOTSTRAP_LOCK_TTL_SECS), - ); - - match recovery_lock.try_acquire().await { - Ok(Some(guard)) => { - warn!("Previous config-processing lock holder appears to have crashed, taking over"); - let result = - process_if_needed_after_lock(config_file, server_config, app_state, conn, prefix) - .await; - drop(guard); - result - } - Ok(None) => { - // Another instance may still be processing config. - // Wait one more bounded period for the explicit completion marker - // before giving up on this instance. - warn!("Config-processing lock still held after timeout, waiting again for completion"); - let completed = wait_for_config_processing_complete(conn, prefix).await?; - - if completed { - info!("Config processing completed by another instance during extended wait"); - Ok(()) - } else { - Err(eyre::eyre!( - "Timed out waiting for config processing and could not acquire recovery lock" - )) - } + server_config.reset_storage_on_start || !is_redis_populated(app_state).await? } - Err(e) => { - warn!( - error = %e, - "Failed to acquire recovery lock for config processing" - ); - Err(e) - } - } -} - -/// Internal function that performs the actual config processing. -async fn execute_config_processing_with_marker( - config_file: &Config, - server_config: &ServerConfig, - app_state: &ThinDataAppState, - conn: &Arc, - prefix: &str, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - set_config_processing_in_progress(conn, prefix).await?; - - let result = execute_config_processing(config_file, server_config, app_state).await; + }; - if result.is_ok() { - set_config_processing_completed(conn, prefix).await?; + if !should_process_config_file { + info!("Skipping config file processing"); + return Ok(()); } - result -} - -/// Internal function that performs the actual config processing work. -async fn execute_config_processing( - config_file: &Config, - server_config: &ServerConfig, - app_state: &ThinDataAppState, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ if server_config.reset_storage_on_start { info!("Resetting storage on start due to server config flag RESET_STORAGE_ON_START = true"); app_state.relayer_repository.drop_all_entries().await?; @@ -686,14 +373,15 @@ where app_state.api_key_repository.drop_all_entries().await?; } - info!("Processing config file"); - process_plugins(config_file, app_state).await?; - process_signers(config_file, app_state).await?; - process_notifications(config_file, app_state).await?; - process_networks(config_file, app_state).await?; - process_relayers(config_file, app_state).await?; - process_api_key(server_config, app_state).await?; - + if should_process_config_file { + info!("Processing config file"); + process_plugins(&config_file, app_state).await?; + process_signers(&config_file, app_state).await?; + process_notifications(&config_file, app_state).await?; + process_networks(&config_file, app_state).await?; + process_relayers(&config_file, app_state).await?; + process_api_key(&server_config, app_state).await?; + } Ok(()) } @@ -1839,18 +1527,6 @@ mod tests { } } - async fn create_test_redis_pool() -> Option> { - let cfg = deadpool_redis::Config::from_url("redis://127.0.0.1:6379"); - let pool = cfg - .builder() - .ok()? - .max_size(16) - .runtime(deadpool_redis::Runtime::Tokio1) - .build() - .ok()?; - Some(Arc::new(pool)) - } - // Helper function to create minimal test config fn create_minimal_test_config() -> Config { Config { @@ -2053,70 +1729,4 @@ mod tests { Ok(()) } - - #[tokio::test] - #[ignore] - async fn test_recover_config_processing_after_timeout_waits_for_extended_completion() { - let conn = create_test_redis_pool() - .await - .expect("Redis connection required"); - let prefix = "test_config_recovery_extended_wait"; - let lock_key = format!("{prefix}:lock:{CONFIG_PROCESSING_LOCK_NAME}"); - let bootstrap_meta_key = format!("{prefix}:bootstrap_meta"); - - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - let _: Result<(), _> = - redis::AsyncCommands::del(&mut conn_clone, &bootstrap_meta_key).await; - } - - let config = create_minimal_test_config(); - let server_config = - create_test_server_config_with_settings(RepositoryStorageType::Redis, false); - let app_state = ThinData(create_test_app_state()); - - let lock = DistributedLock::new( - conn.clone(), - &lock_key, - Duration::from_secs(BOOTSTRAP_LOCK_TTL_SECS), - ); - let guard = lock - .try_acquire() - .await - .expect("Should acquire lock") - .expect("Lock should be available"); - - let conn_for_task = conn.clone(); - let prefix_for_task = prefix.to_string(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(500)).await; - set_config_processing_completed(&conn_for_task, &prefix_for_task) - .await - .expect("Should set config processing completed"); - guard.release().await.expect("Should release lock"); - }); - - let result = recover_config_processing_after_timeout( - &config, - &server_config, - &app_state, - &conn, - prefix, - ) - .await; - - assert!( - result.is_ok(), - "Should succeed when completion is observed during extended wait: {:?}", - result - ); - - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - let _: Result<(), _> = - redis::AsyncCommands::del(&mut conn_clone, &bootstrap_meta_key).await; - } - } } diff --git a/src/bootstrap/initialize_relayers.rs b/src/bootstrap/initialize_relayers.rs index 9924ec2ce..059ef56c0 100644 --- a/src/bootstrap/initialize_relayers.rs +++ b/src/bootstrap/initialize_relayers.rs @@ -2,23 +2,7 @@ //! //! This module contains functions for initializing relayers, ensuring they are //! properly configured and ready for operation. -//! -//! ## Distributed Locking -//! -//! When multiple instances of the relayer service start simultaneously with -//! `DISTRIBUTED_MODE` enabled, this module uses distributed locking to coordinate -//! initialization and prevent duplicate work: -//! -//! - **Global lock**: A single lock is used for the entire initialization process, -//! ensuring only one instance initializes relayers at a time. -//! - **Recent completion check**: Skips initialization if it was recently completed -//! (within the staleness threshold) to handle rolling restarts efficiently. -//! - **Wait for completion**: Instances that don't acquire the lock wait for the -//! initializing instance to complete, then proceed without re-initializing. -//! - **Single-instance mode**: When `DISTRIBUTED_MODE` is disabled (default) or using -//! in-memory storage, locking is skipped since coordination is not needed. use crate::{ - config::ServerConfig, domain::{get_network_relayer, Relayer}, jobs::JobProducerTrait, models::{ @@ -29,23 +13,10 @@ use crate::{ ApiKeyRepositoryTrait, NetworkRepository, PluginRepositoryTrait, RelayerRepository, Repository, TransactionCounterTrait, TransactionRepository, }, - utils::{ - is_global_init_recently_completed, poll_until, set_global_init_completed, DistributedLock, - BOOTSTRAP_LOCK_TTL_SECS, LOCK_POLL_INTERVAL_MS, LOCK_WAIT_MAX_SECS, - }, }; use color_eyre::{eyre::WrapErr, Result}; -use deadpool_redis::Pool; -use std::sync::Arc; -use std::time::Duration; -use tracing::{debug, info, warn}; - -/// Staleness threshold in seconds. Initialization completed within this time is skipped. -/// Set to 5 minutes to prevent redundant initialization on rolling restarts. -const INIT_STALENESS_THRESHOLD_SECS: u64 = 300; - -/// Lock name for global initialization lock. -const GLOBAL_INIT_LOCK_NAME: &str = "relayer_init_global"; +use futures::future::try_join_all; +use tracing::debug; /// Internal function for initializing a relayer using a provided relayer service. /// This allows for easier testing with mocked relayers. @@ -107,339 +78,32 @@ where // Early return for empty list - no work to do if relayers.is_empty() { - info!("No relayers to initialize"); + debug!("No relayers to initialize"); return Ok(()); } - info!(count = relayers.len(), "Initializing relayers"); - - // Check if using persistent storage with distributed coordination - let use_lock = ServerConfig::get_distributed_mode(); - let connection_info = app_state.relayer_repository.connection_info(); - - match (use_lock, connection_info) { - (true, Some((conn, prefix))) => { - // Distributed mode: use locking to coordinate across instances - coordinate_with_distributed_lock(&relayers, &app_state, &conn, &prefix).await - } - _ => { - // Single-instance mode or in-memory storage: skip locking - info!("Initializing relayers without distributed locking"); - run_initialization_batch(&relayers, &app_state).await - } - } -} - -/// Coordinates relayer initialization with a distributed lock for multi-instance deployments. -/// -/// This function handles the coordination logic for distributed initialization: -/// 1. Check if initialization was recently completed (skip if yes) -/// 2. Try to acquire global lock -/// 3. If lock acquired: initialize all relayers and record completion time -/// 4. If lock held by another instance: wait for completion -/// 5. If wait times out: recheck state and attempt recovery (lock holder may have crashed) -/// 6. If lock error: proceed without coordination (graceful degradation) -async fn coordinate_with_distributed_lock( - relayers: &[RelayerRepoModel], - app_state: &ThinDataAppState, - conn: &Arc, - prefix: &str, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - // Step 1: Check if recently completed - match is_global_init_recently_completed(conn, prefix, INIT_STALENESS_THRESHOLD_SECS).await { - Ok(true) => { - info!("Initialization recently completed by another instance, skipping"); - return Ok(()); - } - Ok(false) => {} - Err(e) => { - warn!( - error = %e, - "Failed to check recent initialization status, proceeding with initialization" - ); - } - } - - // Step 2: Try to acquire global lock - let lock_key = format!("{prefix}:lock:{GLOBAL_INIT_LOCK_NAME}"); - let lock = DistributedLock::new( - conn.clone(), - &lock_key, - Duration::from_secs(BOOTSTRAP_LOCK_TTL_SECS), - ); - - let lock_result = lock.try_acquire().await; - - // Handle lock held by another instance - if matches!(&lock_result, Ok(None)) { - info!("Another instance is initializing relayers, waiting for completion"); - let completed = wait_for_initialization_complete(conn, prefix).await?; - - if completed { - return Ok(()); - } - - // Timeout reached — the lock holder may have crashed without completing. - // Recheck: was initialization completed in the final moments? - warn!("Timeout waiting for initialization, rechecking state"); - - if is_global_init_recently_completed(conn, prefix, INIT_STALENESS_THRESHOLD_SECS) - .await - .unwrap_or(false) - { - info!("Initialization completed during timeout window"); - return Ok(()); - } - - // Not completed. Try to acquire the lock and take over initialization. - return recover_after_timeout(relayers, app_state, conn, prefix).await; - } - - // Handle lock error - graceful degradation (early return) - let guard = match lock_result { - Ok(Some(g)) => g, - Err(e) => { - warn!( - error = %e, - "Failed to acquire distributed lock, proceeding without coordination" - ); - return run_initialization_batch(relayers, app_state).await; - } - Ok(None) => unreachable!(), // Already handled above - }; - - // Lock acquired - proceed with initialization - info!( - count = relayers.len(), - "Acquired initialization lock, initializing relayers" - ); - - let result = run_initialization_batch(relayers, app_state).await; - - // Record completion time only on success - if result.is_ok() { - if let Err(e) = set_global_init_completed(conn, prefix).await { - warn!(error = %e, "Failed to record initialization completion time"); - } - } - - drop(guard); - result -} - -/// Waits for another instance to complete initialization. -/// -/// Polls periodically until: -/// - Initialization is completed (detected via recent completion timestamp) → returns `Ok(true)` -/// - Timeout is reached without completion detected → returns `Ok(false)` -async fn wait_for_initialization_complete(conn: &Arc, prefix: &str) -> Result { - let max_wait = Duration::from_secs(LOCK_WAIT_MAX_SECS); - let poll_interval = Duration::from_millis(LOCK_POLL_INTERVAL_MS); - - // Clone values for the closure - let conn = conn.clone(); - let prefix = prefix.to_string(); - - poll_until( - || is_global_init_recently_completed(&conn, &prefix, INIT_STALENESS_THRESHOLD_SECS), - max_wait, - poll_interval, - "initialization", - ) - .await -} - -/// Attempts to recover after a wait timeout by acquiring the lock and initializing. -/// -/// This handles the case where the lock holder crashed or errored without completing -/// initialization. After timeout: -/// - If the lock can be acquired (previous holder's TTL expired): take over and initialize -/// - If the lock is still held (another instance is legitimately running): wait one more -/// bounded period for completion, then initialize as last resort -/// - If Redis errors: graceful degradation (initialize without coordination) -async fn recover_after_timeout( - relayers: &[RelayerRepoModel], - app_state: &ThinDataAppState, - conn: &Arc, - prefix: &str, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - let lock_key = format!("{prefix}:lock:{GLOBAL_INIT_LOCK_NAME}"); - let recovery_lock = DistributedLock::new( - conn.clone(), - &lock_key, - Duration::from_secs(BOOTSTRAP_LOCK_TTL_SECS), - ); - - match recovery_lock.try_acquire().await { - Ok(Some(guard)) => { - // Lock expired (holder crashed) — we take over - warn!( - count = relayers.len(), - "Previous lock holder appears to have crashed, taking over initialization" - ); - let result = run_initialization_batch(relayers, app_state).await; - if result.is_ok() { - if let Err(e) = set_global_init_completed(conn, prefix).await { - warn!(error = %e, "Failed to record initialization completion time"); - } - } - drop(guard); - result - } - Ok(None) => { - // Lock still held — another instance is legitimately running. - // Wait one more bounded period for completion instead of failing. - warn!("Lock still held by another instance after timeout, waiting for completion"); - let completed = wait_for_initialization_complete(conn, prefix).await?; + debug!(count = relayers.len(), "Initializing relayers concurrently"); - if completed { - info!("Initialization completed by another instance during extended wait"); - Ok(()) - } else { - // Extended wait also timed out (~260s total). Proceed with initialization - // as a last resort rather than failing — duplicate side effects (notifications, - // jobs) are a minor cost compared to no instance initializing at all. - warn!("Extended wait also timed out, proceeding with initialization"); - let result = run_initialization_batch(relayers, app_state).await; - if result.is_ok() { - if let Err(e) = set_global_init_completed(conn, prefix).await { - warn!(error = %e, "Failed to record initialization completion time"); - } - } - result - } - } - Err(e) => { - // Redis error — graceful degradation - warn!( - error = %e, - "Failed to check lock after timeout, attempting initialization without coordination" - ); - run_initialization_batch(relayers, app_state).await - } - } -} - -/// Runs the batch initialization of all relayers concurrently. -async fn run_initialization_batch( - relayers: &[RelayerRepoModel], - app_state: &ThinDataAppState, -) -> Result<()> -where - J: JobProducerTrait + Send + Sync + 'static, - RR: RelayerRepository + Repository + Send + Sync + 'static, - TR: TransactionRepository + Repository + Send + Sync + 'static, - NR: NetworkRepository + Repository + Send + Sync + 'static, - NFR: Repository + Send + Sync + 'static, - SR: Repository + Send + Sync + 'static, - TCR: TransactionCounterTrait + Send + Sync + 'static, - PR: PluginRepositoryTrait + Send + Sync + 'static, - AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, -{ - let futures = relayers.iter().map(|relayer| { + let relayer_futures = relayers.iter().map(|relayer| { let app_state = app_state.clone(); - let relayer_id = relayer.id.clone(); - - async move { - let result = initialize_relayer(relayer_id.clone(), app_state).await; - (relayer_id, result) - } + async move { initialize_relayer(relayer.id.clone(), app_state).await } }); - let results = futures::future::join_all(futures).await; - - // Count and report results - let succeeded = results.iter().filter(|(_, r)| r.is_ok()).count(); - let failed = results.iter().filter(|(_, r)| r.is_err()).count(); + try_join_all(relayer_futures) + .await + .wrap_err("Failed to initialize relayers")?; - info!( - succeeded = succeeded, - failed = failed, - "Relayer initialization completed" + debug!( + count = relayers.len(), + "All relayers initialized successfully" ); - - // Collect failures and return error if any - if failed > 0 { - let failures: Vec = results - .into_iter() - .filter_map(|(id, r)| r.err().map(|e| format!("{id}: {e}"))) - .collect(); - - return Err(eyre::eyre!( - "Failed to initialize {} relayer(s): {}", - failed, - failures.join("; ") - )); - } - Ok(()) } #[cfg(test)] mod tests { use super::*; - use crate::{ - jobs::MockJobProducerTrait, - models::{AppState, RepositoryError}, - repositories::{ - ApiKeyRepositoryStorage, MockRelayerRepository, NetworkRepositoryStorage, - NotificationRepositoryStorage, PluginRepositoryStorage, SignerRepositoryStorage, - TransactionCounterRepositoryStorage, TransactionRepositoryStorage, - }, - utils::mocks::mockutils::{create_mock_app_state, create_mock_relayer}, - }; - use actix_web::web::ThinData; - use std::sync::Arc; - - fn create_app_state_with_mock_relayer_repo( - mock_relayer_repo: MockRelayerRepository, - ) -> AppState< - MockJobProducerTrait, - MockRelayerRepository, - TransactionRepositoryStorage, - NetworkRepositoryStorage, - NotificationRepositoryStorage, - SignerRepositoryStorage, - TransactionCounterRepositoryStorage, - PluginRepositoryStorage, - ApiKeyRepositoryStorage, - > { - AppState { - relayer_repository: Arc::new(mock_relayer_repo), - transaction_repository: Arc::new(TransactionRepositoryStorage::new_in_memory()), - signer_repository: Arc::new(SignerRepositoryStorage::new_in_memory()), - notification_repository: Arc::new(NotificationRepositoryStorage::new_in_memory()), - network_repository: Arc::new(NetworkRepositoryStorage::new_in_memory()), - transaction_counter_store: Arc::new( - TransactionCounterRepositoryStorage::new_in_memory(), - ), - job_producer: Arc::new(MockJobProducerTrait::new()), - plugin_repository: Arc::new(PluginRepositoryStorage::new_in_memory()), - api_key_repository: Arc::new(ApiKeyRepositoryStorage::new_in_memory()), - } - } + use crate::utils::mocks::mockutils::create_mock_relayer; #[test] fn test_get_relayer_ids_with_empty_list() { @@ -635,564 +299,4 @@ mod tests { "Second relayer should initialize successfully" ); } - - // Tests for constants - #[test] - fn test_lock_ttl_is_reasonable() { - // Lock TTL should be at least 60 seconds to handle slow initializations - assert!( - BOOTSTRAP_LOCK_TTL_SECS >= 60, - "Lock TTL should be at least 60 seconds" - ); - // But not too long (more than 10 minutes would be excessive) - assert!( - BOOTSTRAP_LOCK_TTL_SECS <= 600, - "Lock TTL should not exceed 10 minutes" - ); - } - - #[test] - fn test_staleness_threshold_is_reasonable() { - // Staleness threshold should be at least 60 seconds - assert!( - INIT_STALENESS_THRESHOLD_SECS >= 60, - "Staleness threshold should be at least 60 seconds" - ); - // But not too long (more than 1 hour would be excessive) - assert!( - INIT_STALENESS_THRESHOLD_SECS <= 3600, - "Staleness threshold should not exceed 1 hour" - ); - } - - #[test] - fn test_wait_max_duration_exceeds_lock_ttl() { - // Wait duration should be longer than lock TTL to handle edge cases - assert!( - LOCK_WAIT_MAX_SECS > BOOTSTRAP_LOCK_TTL_SECS, - "Wait duration should exceed lock TTL" - ); - } - - #[test] - fn test_poll_interval_is_reasonable() { - // Poll interval should be at least 100ms to avoid excessive polling - assert!( - LOCK_POLL_INTERVAL_MS >= 100, - "Poll interval should be at least 100ms" - ); - // But not too long (more than 5 seconds would be slow) - assert!( - LOCK_POLL_INTERVAL_MS <= 5000, - "Poll interval should not exceed 5 seconds" - ); - } - - // Tests for get_relayer_ids_to_initialize edge cases - #[test] - fn test_get_relayer_ids_preserves_order() { - let relayers = vec![ - create_mock_relayer("z-relayer".to_string(), false), - create_mock_relayer("a-relayer".to_string(), false), - create_mock_relayer("m-relayer".to_string(), false), - ]; - - let ids = get_relayer_ids_to_initialize(&relayers); - - // Should preserve insertion order, not sort - assert_eq!(ids[0], "z-relayer"); - assert_eq!(ids[1], "a-relayer"); - assert_eq!(ids[2], "m-relayer"); - } - - #[test] - fn test_get_relayer_ids_with_special_characters() { - let relayers = vec![ - create_mock_relayer("relayer-with-dashes".to_string(), false), - create_mock_relayer("relayer_with_underscores".to_string(), false), - create_mock_relayer("relayer.with.dots".to_string(), false), - ]; - - let ids = get_relayer_ids_to_initialize(&relayers); - - assert_eq!(ids.len(), 3); - assert!(ids.contains(&"relayer-with-dashes".to_string())); - assert!(ids.contains(&"relayer_with_underscores".to_string())); - assert!(ids.contains(&"relayer.with.dots".to_string())); - } - - #[test] - fn test_get_relayer_ids_with_large_list() { - let relayers: Vec = (0..100) - .map(|i| create_mock_relayer(format!("relayer-{:03}", i), false)) - .collect(); - - let ids = get_relayer_ids_to_initialize(&relayers); - - assert_eq!(ids.len(), 100); - assert_eq!(ids[0], "relayer-000"); - assert_eq!(ids[99], "relayer-099"); - } - - // Test error message formatting - #[tokio::test] - async fn test_initialize_relayer_with_service_error_includes_relayer_id() { - use crate::domain::MockRelayer; - use crate::models::RelayerError; - - let mut mock_relayer = MockRelayer::new(); - mock_relayer - .expect_initialize_relayer() - .times(1) - .returning(|| { - Box::pin(async { - Err(RelayerError::NetworkConfiguration("bad config".to_string())) - }) - }); - - let result = initialize_relayer_with_service("my-special-relayer-id", &mock_relayer).await; - - assert!(result.is_err()); - let err_str = result.unwrap_err().to_string(); - assert!( - err_str.contains("my-special-relayer-id"), - "Error should contain relayer ID, got: {}", - err_str - ); - } - - #[tokio::test] - async fn test_initialize_relayer_with_service_provider_error() { - use crate::domain::MockRelayer; - use crate::models::RelayerError; - - let mut mock_relayer = MockRelayer::new(); - mock_relayer - .expect_initialize_relayer() - .times(1) - .returning(|| { - Box::pin(async { Err(RelayerError::ProviderError("provider failed".to_string())) }) - }); - - let result = initialize_relayer_with_service("test-relayer", &mock_relayer).await; - assert!(result.is_err(), "Should fail for ProviderError"); - } - - #[tokio::test] - async fn test_initialize_relayer_with_service_network_config_error() { - use crate::domain::MockRelayer; - use crate::models::RelayerError; - - let mut mock_relayer = MockRelayer::new(); - mock_relayer - .expect_initialize_relayer() - .times(1) - .returning(|| { - Box::pin(async { - Err(RelayerError::NetworkConfiguration( - "network config error".to_string(), - )) - }) - }); - - let result = initialize_relayer_with_service("test-relayer", &mock_relayer).await; - assert!( - result.is_err(), - "Should fail for NetworkConfiguration error" - ); - } - - // ============================================================================ - // Integration tests for run_initialization_batch, coordinate_with_distributed_lock, - // wait_for_initialization_complete, and initialize_relayers - // ============================================================================ - - /// Helper to create a Redis connection pool for integration tests. - async fn create_test_redis_pool() -> Option> { - let cfg = deadpool_redis::Config::from_url("redis://127.0.0.1:6379"); - let pool = cfg - .builder() - .ok()? - .max_size(16) - .runtime(deadpool_redis::Runtime::Tokio1) - .build() - .ok()?; - Some(Arc::new(pool)) - } - - fn create_unreachable_redis_pool() -> Arc { - let cfg = deadpool_redis::Config::from_url("redis://127.0.0.1:1"); - let pool = cfg - .builder() - .expect("should create deadpool builder") - .max_size(1) - .runtime(deadpool_redis::Runtime::Tokio1) - .build() - .expect("should build deadpool"); - Arc::new(pool) - } - - // --- Tests for run_initialization_batch --- - - #[tokio::test] - async fn test_run_initialization_batch_empty_list() { - let app_state = create_mock_app_state(None, None, None, None, None, None).await; - let thin_state = ThinData(app_state); - - let relayers: Vec = vec![]; - let result = run_initialization_batch(&relayers, &thin_state).await; - - assert!( - result.is_ok(), - "Should succeed with empty relayer list: {:?}", - result - ); - } - - #[tokio::test] - async fn test_run_initialization_batch_handles_failures() { - let relayers = vec![create_mock_relayer("failing-relayer".to_string(), false)]; - - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let result = run_initialization_batch(&relayers, &thin_state).await; - - assert!(result.is_err(), "Should fail due to missing signer"); - } - - #[tokio::test] - async fn test_run_initialization_batch_concurrent_execution() { - let relayers: Vec = (0..5) - .map(|i| create_mock_relayer(format!("concurrent-relayer-{}", i), false)) - .collect(); - - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - // This will fail because signers aren't configured, but it tests concurrent execution - let result = run_initialization_batch(&relayers, &thin_state).await; - - assert!(result.is_err(), "Should fail due to missing signers"); - // The error message should mention the failed relayers - let err_str = result.unwrap_err().to_string(); - assert!( - err_str.contains("Failed to initialize"), - "Error should mention initialization failure" - ); - } - - #[tokio::test] - async fn test_run_initialization_batch_reports_each_failed_relayer_id() { - let relayers = vec![ - create_mock_relayer("batch-relayer-1".to_string(), false), - create_mock_relayer("batch-relayer-2".to_string(), false), - ]; - - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let result = run_initialization_batch(&relayers, &thin_state).await; - - assert!(result.is_err(), "Should fail due to missing signers"); - - let err = result.unwrap_err().to_string(); - assert!(err.contains("Failed to initialize 2 relayer(s)")); - assert!(err.contains("batch-relayer-1")); - assert!(err.contains("batch-relayer-2")); - } - - // --- Tests for recover_after_timeout --- - - #[tokio::test] - async fn test_recover_after_timeout_gracefully_degrades_when_redis_errors() { - let relayers = vec![create_mock_relayer( - "recovery-error-relayer".to_string(), - false, - )]; - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - let unreachable_pool = create_unreachable_redis_pool(); - - let result = - recover_after_timeout(&relayers, &thin_state, &unreachable_pool, "recover_err").await; - - assert!(result.is_err(), "Should fall back to initialization batch"); - let err = result.unwrap_err().to_string(); - assert!(err.contains("Failed to initialize")); - assert!(err.contains("recovery-error-relayer")); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_recover_after_timeout_acquires_lock_and_attempts_initialization() { - let conn = create_test_redis_pool() - .await - .expect("Redis connection required"); - let relayers = vec![create_mock_relayer( - "recovery-lock-relayer".to_string(), - false, - )]; - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - let prefix = "test_recover_after_timeout_acquire"; - - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let lock_key = format!("{prefix}:lock:{GLOBAL_INIT_LOCK_NAME}"); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - } - - let result = recover_after_timeout(&relayers, &thin_state, &conn, prefix).await; - - assert!( - result.is_err(), - "Should attempt initialization after taking over the lock" - ); - - let is_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check completion"); - assert!( - !is_recent, - "Should not record completion time when initialization fails" - ); - - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let lock_key = format!("{prefix}:lock:{GLOBAL_INIT_LOCK_NAME}"); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - } - } - - // --- Tests for coordinate_with_distributed_lock (requires Redis) --- - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_coordinate_with_distributed_lock_skips_when_recently_completed() { - let conn = create_test_redis_pool() - .await - .expect("Redis connection required"); - - let relayers = vec![create_mock_relayer( - "global-lock-relayer".to_string(), - false, - )]; - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let prefix = "test_global_skip_recent"; - - // Set completion time to simulate recent initialization - set_global_init_completed(&conn, prefix) - .await - .expect("Should set completion time"); - - let result = coordinate_with_distributed_lock(&relayers, &thin_state, &conn, prefix).await; - - // Should succeed because it skips (recently completed) - assert!( - result.is_ok(), - "Should skip initialization when recently completed: {:?}", - result - ); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{}:relayer_sync_meta", prefix); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_coordinate_with_distributed_lock_acquires_lock() { - let conn = create_test_redis_pool() - .await - .expect("Redis connection required"); - - let relayers = vec![create_mock_relayer( - "lock-acquire-relayer".to_string(), - false, - )]; - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let prefix = "test_global_acquire_lock"; - - // Clear any existing state - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{}:relayer_sync_meta", prefix); - let lock_key = format!("{}:lock:{}", prefix, GLOBAL_INIT_LOCK_NAME); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - } - - let result = coordinate_with_distributed_lock(&relayers, &thin_state, &conn, prefix).await; - - // Will fail because signer isn't configured, but lock should have been acquired - assert!( - result.is_err(), - "Should fail due to missing signer configuration" - ); - - // Verify completion time was NOT set (because initialization failed) - let is_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check completion"); - assert!(!is_recent, "Should NOT record completion time on failure"); - - // Cleanup - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{}:relayer_sync_meta", prefix); - let lock_key = format!("{}:lock:{}", prefix, GLOBAL_INIT_LOCK_NAME); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - } - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_coordinate_with_distributed_lock_waits_when_lock_held() { - let conn = create_test_redis_pool() - .await - .expect("Redis connection required"); - - let relayers = vec![create_mock_relayer("wait-relayer".to_string(), false)]; - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let prefix = "test_global_wait_lock"; - let hash_key = format!("{}:relayer_sync_meta", prefix); - let lock_key = format!("{}:lock:{}", prefix, GLOBAL_INIT_LOCK_NAME); - - // Clear any existing state - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - } - - // Acquire lock to simulate another instance initializing - let lock = DistributedLock::new(conn.clone(), &lock_key, Duration::from_secs(5)); - let guard = lock - .try_acquire() - .await - .expect("Should acquire lock") - .expect("Lock should be available"); - - // Spawn task to release lock and set completion after a short delay - let conn_for_task = conn.clone(); - let prefix_for_task = prefix.to_string(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(500)).await; - set_global_init_completed(&conn_for_task, &prefix_for_task) - .await - .expect("Should set completion"); - guard.release().await.expect("Should release lock"); - }); - - // This should wait and then succeed (because completion will be set) - let result = coordinate_with_distributed_lock(&relayers, &thin_state, &conn, prefix).await; - - assert!( - result.is_ok(), - "Should succeed after waiting for completion: {:?}", - result - ); - - // Cleanup - { - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &hash_key).await; - let _: Result<(), _> = redis::AsyncCommands::del(&mut conn_clone, &lock_key).await; - } - } - - // --- Tests for initialize_relayers main function --- - - #[tokio::test] - async fn test_initialize_relayers_empty_list() { - let app_state = create_mock_app_state(None, None, None, None, None, None).await; - let thin_state = ThinData(app_state); - - let result = initialize_relayers(thin_state).await; - - assert!( - result.is_ok(), - "Should succeed with empty relayer list: {:?}", - result - ); - } - - #[tokio::test] - async fn test_initialize_relayers_uses_in_memory_path() { - let relayers = vec![create_mock_relayer("inmem-relayer".to_string(), false)]; - let app_state = - create_mock_app_state(None, Some(relayers.clone()), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let result = initialize_relayers(thin_state).await; - - assert!( - result.is_err(), - "Should fail due to missing signer configuration" - ); - } - - #[tokio::test] - async fn test_initialize_relayer_returns_error_when_relayer_is_missing() { - let app_state = create_mock_app_state(None, None, None, None, None, None).await; - let thin_state = ThinData(app_state); - - let result = initialize_relayer("missing-relayer".to_string(), thin_state).await; - - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("missing-relayer")); - } - - #[tokio::test] - async fn test_initialize_relayer_returns_error_when_signer_is_missing() { - let relayers = vec![create_mock_relayer("signerless-relayer".to_string(), false)]; - let app_state = create_mock_app_state(None, Some(relayers), None, None, None, None).await; - let thin_state = ThinData(app_state); - - let result = initialize_relayer("signerless-relayer".to_string(), thin_state).await; - - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("test")); - } - - #[tokio::test] - async fn test_initialize_relayers_propagates_list_all_error() { - let mut mock_relayer_repo = MockRelayerRepository::new(); - mock_relayer_repo.expect_list_all().times(1).returning(|| { - Err(RepositoryError::ConnectionError( - "relayer repository unavailable".to_string(), - )) - }); - - let thin_state = ThinData(create_app_state_with_mock_relayer_repo(mock_relayer_repo)); - - let result = initialize_relayers(thin_state).await; - - assert!(result.is_err()); - assert!(result - .unwrap_err() - .to_string() - .contains("relayer repository unavailable")); - } } diff --git a/src/repositories/api_key/api_key_redis.rs b/src/repositories/api_key/api_key_redis.rs index a9a3c8e3a..561d8f005 100644 --- a/src/repositories/api_key/api_key_redis.rs +++ b/src/repositories/api_key/api_key_redis.rs @@ -330,14 +330,14 @@ impl ApiKeyRepositoryTrait for RedisApiKeyRepository { .await?; let plugin_list_key = self.api_key_list_key(); - debug!("Checking if API key entries exist"); + debug!("Checking if plugin entries exist"); let exists: bool = conn .exists(&plugin_list_key) .await .map_err(|e| self.map_redis_error(e, "has_entries_check"))?; - debug!("API key entries exist: {}", exists); + debug!("Plugin entries exist: {}", exists); Ok(exists) } @@ -345,18 +345,18 @@ impl ApiKeyRepositoryTrait for RedisApiKeyRepository { let mut conn = self .get_connection(self.connections.primary(), "drop_all_entries") .await?; - let api_key_list_key = self.api_key_list_key(); + let plugin_list_key = self.api_key_list_key(); - debug!("Dropping all API key entries"); + debug!("Dropping all plugin entries"); - // Get all API key IDs first - let api_key_ids: Vec = conn - .smembers(&api_key_list_key) + // Get all plugin IDs first + let plugin_ids: Vec = conn + .smembers(&plugin_list_key) .await .map_err(|e| self.map_redis_error(e, "drop_all_entries_get_ids"))?; - if api_key_ids.is_empty() { - debug!("No API key entries to drop"); + if plugin_ids.is_empty() { + debug!("No plugin entries to drop"); return Ok(()); } @@ -364,20 +364,20 @@ impl ApiKeyRepositoryTrait for RedisApiKeyRepository { let mut pipe = redis::pipe(); pipe.atomic(); - // 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 all individual plugin entries + for plugin_id in &plugin_ids { + let plugin_key = self.api_key_key(plugin_id); + pipe.del(&plugin_key); } - // Delete the API key list key - pipe.del(&api_key_list_key); + // Delete the plugin list key + pipe.del(&plugin_list_key); pipe.exec_async(&mut conn) .await .map_err(|e| self.map_redis_error(e, "drop_all_entries_pipeline"))?; - debug!("Dropped {} API key entries", api_key_ids.len()); + debug!("Dropped {} plugin entries", plugin_ids.len()); Ok(()) } } diff --git a/src/repositories/relayer/mod.rs b/src/repositories/relayer/mod.rs index 8067413ec..f0f428d60 100644 --- a/src/repositories/relayer/mod.rs +++ b/src/repositories/relayer/mod.rs @@ -34,7 +34,6 @@ use crate::{ utils::RedisConnections, }; use async_trait::async_trait; -use deadpool_redis::Pool; use std::sync::Arc; #[async_trait] @@ -68,19 +67,6 @@ pub trait RelayerRepository: Repository + 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, String)> { - None - } } #[cfg(test)] @@ -110,7 +96,6 @@ mockall::mock! { async fn disable_relayer(&self, relayer_id: String, reason: DisabledReason) -> Result; async fn update_policy(&self, id: String, policy: RelayerNetworkPolicy) -> Result; fn is_persistent_storage(&self) -> bool; - fn connection_info(&self) -> Option<(Arc, String)>; } } @@ -135,24 +120,6 @@ 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, String)> { - match self { - RelayerRepositoryStorage::InMemory(_) => None, - RelayerRepositoryStorage::Redis(repo) => { - Some((repo.connections.primary().clone(), repo.key_prefix.clone())) - } - } - } } impl Default for RelayerRepositoryStorage { @@ -318,15 +285,6 @@ impl RelayerRepository for RelayerRepositoryStorage { RelayerRepositoryStorage::Redis(_) => true, } } - - fn connection_info(&self) -> Option<(Arc, String)> { - match self { - RelayerRepositoryStorage::InMemory(_) => None, - RelayerRepositoryStorage::Redis(repo) => { - Some((repo.connections.primary().clone(), repo.key_prefix.clone())) - } - } - } } #[cfg(test)] @@ -541,38 +499,4 @@ 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, String)> = storage.connection_info(); - assert!(result.is_none()); - } } diff --git a/src/repositories/relayer/relayer_in_memory.rs b/src/repositories/relayer/relayer_in_memory.rs index 0f0981ad2..b9db30317 100644 --- a/src/repositories/relayer/relayer_in_memory.rs +++ b/src/repositories/relayer/relayer_in_memory.rs @@ -166,8 +166,6 @@ impl RelayerRepository for InMemoryRelayerRepository { fn is_persistent_storage(&self) -> bool { false } - - // Uses the trait default implementation which returns None } #[async_trait] diff --git a/src/repositories/relayer/relayer_redis.rs b/src/repositories/relayer/relayer_redis.rs index 49626ef9e..e402e9993 100644 --- a/src/repositories/relayer/relayer_redis.rs +++ b/src/repositories/relayer/relayer_redis.rs @@ -560,10 +560,6 @@ impl RelayerRepository for RedisRelayerRepository { fn is_persistent_storage(&self) -> bool { true } - - fn connection_info(&self) -> Option<(Arc, String)> { - Some((self.connections.primary().clone(), self.key_prefix.clone())) - } } #[cfg(test)] diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 0e2ced380..2f30ace46 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -8,9 +8,6 @@ pub use key::*; mod auth; pub use auth::*; -mod polling; -pub use polling::*; - mod time; pub use time::*; diff --git a/src/utils/polling.rs b/src/utils/polling.rs deleted file mode 100644 index df07415c2..000000000 --- a/src/utils/polling.rs +++ /dev/null @@ -1,192 +0,0 @@ -use std::future::Future; -use std::time::Duration; - -use color_eyre::Result; -use tracing::{debug, warn}; - -/// Polls until a condition is met or timeout is reached. -/// -/// This helper provides a reusable abstraction for waiting on async conditions -/// with configurable timeout and polling interval. -/// -/// # Arguments -/// * `check` - Closure that returns `Ok(true)` when condition is met, `Ok(false)` to continue polling -/// * `max_wait` - Maximum time to wait before giving up -/// * `poll_interval` - Time to sleep between polls -/// * `operation_name` - Name of the operation for logging -/// -/// # Returns -/// * `Ok(true)` - Condition was met within timeout -/// * `Ok(false)` - Timeout reached without condition being met (errors are logged and polling continues) -pub async fn poll_until( - check: F, - max_wait: Duration, - poll_interval: Duration, - operation_name: &str, -) -> Result -where - F: Fn() -> Fut, - Fut: Future>, -{ - let start = std::time::Instant::now(); - - loop { - match check().await { - Ok(true) => { - debug!("{} completed", operation_name); - return Ok(true); - } - Ok(false) => {} - Err(e) => { - warn!(error = %e, "Error checking {} status while waiting", operation_name); - } - } - - if start.elapsed() > max_wait { - warn!( - "Timed out waiting for {} to complete, proceeding anyway", - operation_name - ); - return Ok(false); - } - - tokio::time::sleep(poll_interval).await; - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::atomic::{AtomicU32, Ordering}; - use std::sync::Arc; - - #[tokio::test] - async fn test_poll_until_condition_met_immediately() { - let result = poll_until( - || async { Ok(true) }, - Duration::from_millis(100), - Duration::from_millis(10), - "immediate_test", - ) - .await; - - assert!(result.is_ok()); - assert!(result.unwrap()); - } - - #[tokio::test] - async fn test_poll_until_condition_met_after_multiple_polls() { - let poll_count = Arc::new(AtomicU32::new(0)); - let poll_count_clone = Arc::clone(&poll_count); - - let result = poll_until( - move || { - let count = poll_count_clone.fetch_add(1, Ordering::SeqCst); - async move { - // Return true on the 3rd poll (count == 2) - Ok(count >= 2) - } - }, - Duration::from_secs(1), - Duration::from_millis(10), - "delayed_condition_test", - ) - .await; - - assert!(result.is_ok()); - assert!(result.unwrap()); - assert!(poll_count.load(Ordering::SeqCst) >= 3); - } - - #[tokio::test] - async fn test_poll_until_timeout_reached() { - let result = poll_until( - || async { Ok(false) }, - Duration::from_millis(50), - Duration::from_millis(10), - "timeout_test", - ) - .await; - - assert!(result.is_ok()); - assert!(!result.unwrap()); - } - - #[tokio::test] - async fn test_poll_until_continues_polling_after_errors() { - let poll_count = Arc::new(AtomicU32::new(0)); - let poll_count_clone = Arc::clone(&poll_count); - - let result = poll_until( - move || { - let count = poll_count_clone.fetch_add(1, Ordering::SeqCst); - async move { - if count < 2 { - // Return error on first two polls - Err(color_eyre::eyre::eyre!("temporary error")) - } else { - // Return success on 3rd poll - Ok(true) - } - } - }, - Duration::from_secs(1), - Duration::from_millis(10), - "error_recovery_test", - ) - .await; - - assert!(result.is_ok()); - assert!(result.unwrap()); - assert!(poll_count.load(Ordering::SeqCst) >= 3); - } - - #[tokio::test] - async fn test_poll_until_timeout_after_persistent_errors() { - let poll_count = Arc::new(AtomicU32::new(0)); - let poll_count_clone = Arc::clone(&poll_count); - - let result = poll_until( - move || { - poll_count_clone.fetch_add(1, Ordering::SeqCst); - async { Err(color_eyre::eyre::eyre!("persistent error")) } - }, - Duration::from_millis(50), - Duration::from_millis(10), - "persistent_error_test", - ) - .await; - - // Should timeout (return Ok(false)) since errors don't stop polling - assert!(result.is_ok()); - assert!(!result.unwrap()); - // Should have polled multiple times - assert!(poll_count.load(Ordering::SeqCst) >= 2); - } - - #[tokio::test] - async fn test_poll_until_respects_poll_interval() { - let start = std::time::Instant::now(); - let poll_count = Arc::new(AtomicU32::new(0)); - let poll_count_clone = Arc::clone(&poll_count); - - let result = poll_until( - move || { - let count = poll_count_clone.fetch_add(1, Ordering::SeqCst); - async move { Ok(count >= 3) } - }, - Duration::from_secs(1), - Duration::from_millis(50), - "interval_test", - ) - .await; - - let elapsed = start.elapsed(); - - assert!(result.is_ok()); - assert!(result.unwrap()); - // With 50ms interval and 4 polls (0, 1, 2, 3), we expect at least 150ms - // (3 sleeps between 4 polls) - assert!(elapsed >= Duration::from_millis(100)); - } -} diff --git a/src/utils/redis.rs b/src/utils/redis.rs index 472033450..f5e1f4b9f 100644 --- a/src/utils/redis.rs +++ b/src/utils/redis.rs @@ -7,21 +7,6 @@ use tracing::{debug, info, warn}; use crate::config::ServerConfig; -// ============================================================================ -// Shared Timing Constants for Bootstrap Operations -// ============================================================================ - -/// Default lock TTL for bootstrap operations (2 minutes). -/// Used as a safety net for crashes during initialization or config processing. -pub const BOOTSTRAP_LOCK_TTL_SECS: u64 = 120; - -/// Max wait time when another instance holds the lock. -/// Set slightly longer than lock TTL to handle edge cases. -pub const LOCK_WAIT_MAX_SECS: u64 = 130; - -/// Polling interval when waiting for completion. -pub const LOCK_POLL_INTERVAL_MS: u64 = 500; - /// Holds separate connection pools for read and write operations. /// /// This struct enables optimization for Redis deployments with read replicas, @@ -391,320 +376,6 @@ fn generate_lock_value() -> String { format!("{process_id}:{timestamp}:{counter}") } -// ============================================================================ -// Relayer Sync Metadata Functions -// ============================================================================ -// -// These functions track when relayers were last synchronized/initialized. -// This allows multiple instances to skip redundant initialization when -// a relayer was recently synced by another instance. - -/// The Redis hash key suffix for storing relayer sync metadata. -const RELAYER_SYNC_META_KEY: &str = "relayer_sync_meta"; - -/// The hash field for global initialization completion timestamp. -const GLOBAL_INIT_FIELD: &str = "global:init_completed"; - -/// The Redis hash key suffix for storing bootstrap process metadata. -const BOOTSTRAP_META_KEY: &str = "bootstrap_meta"; - -/// Hash field storing config processing status. -const CONFIG_PROCESSING_STATUS_FIELD: &str = "config_processing:status"; - -/// Hash field storing config processing start timestamp. -const CONFIG_PROCESSING_STARTED_AT_FIELD: &str = "config_processing:started_at"; - -/// Hash field storing config processing completion timestamp. -const CONFIG_PROCESSING_COMPLETED_AT_FIELD: &str = "config_processing:completed_at"; - -const CONFIG_PROCESSING_STATUS_IN_PROGRESS: &str = "in_progress"; -const CONFIG_PROCESSING_STATUS_COMPLETED: &str = "completed"; - -/// Marks config processing as in progress. -/// -/// This explicit marker is used by distributed bootstrap coordination so -/// waiters do not infer completion from partially written repository state. -pub async fn set_config_processing_in_progress(pool: &Arc, prefix: &str) -> Result<()> { - use chrono::Utc; - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}"); - let timestamp = Utc::now().timestamp(); - - conn.hset_multiple::<_, _, _, ()>( - &hash_key, - &[ - ( - CONFIG_PROCESSING_STATUS_FIELD, - CONFIG_PROCESSING_STATUS_IN_PROGRESS, - ), - (CONFIG_PROCESSING_STARTED_AT_FIELD, ×tamp.to_string()), - ], - ) - .await - .map_err(|e| eyre::eyre!("Failed to set config processing in progress: {}", e))?; - - conn.hdel::<_, _, ()>(&hash_key, CONFIG_PROCESSING_COMPLETED_AT_FIELD) - .await - .map_err(|e| eyre::eyre!("Failed to clear config processing completion time: {}", e))?; - - debug!( - timestamp = %timestamp, - "recorded config processing in-progress marker" - ); - - Ok(()) -} - -/// Marks config processing as completed. -pub async fn set_config_processing_completed(pool: &Arc, prefix: &str) -> Result<()> { - use chrono::Utc; - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}"); - let timestamp = Utc::now().timestamp(); - - conn.hset_multiple::<_, _, _, ()>( - &hash_key, - &[ - ( - CONFIG_PROCESSING_STATUS_FIELD, - CONFIG_PROCESSING_STATUS_COMPLETED, - ), - (CONFIG_PROCESSING_COMPLETED_AT_FIELD, ×tamp.to_string()), - ], - ) - .await - .map_err(|e| eyre::eyre!("Failed to set config processing completed: {}", e))?; - - debug!( - timestamp = %timestamp, - "recorded config processing completed marker" - ); - - Ok(()) -} - -/// Returns whether config processing has been explicitly marked as completed. -pub async fn is_config_processing_completed(pool: &Arc, prefix: &str) -> Result { - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{BOOTSTRAP_META_KEY}"); - - let status: Option = conn - .hget(&hash_key, CONFIG_PROCESSING_STATUS_FIELD) - .await - .map_err(|e| eyre::eyre!("Failed to get config processing status: {}", e))?; - - Ok(matches!( - status.as_deref(), - Some(CONFIG_PROCESSING_STATUS_COMPLETED) - )) -} - -/// Sets the last sync timestamp for a relayer to the current time. -/// -/// This should be called after a relayer has been successfully initialized -/// to record when the initialization occurred. -/// -/// # Arguments -/// * `conn` - Redis connection manager -/// * `prefix` - Key prefix for multi-tenant support -/// * `relayer_id` - The relayer's unique identifier -/// -/// # Redis Key Format -/// Hash key: `{prefix}:relayer_sync_meta` -/// Hash field: `{relayer_id}:last_sync` -/// Value: Unix timestamp in seconds -pub async fn set_relayer_last_sync(pool: &Arc, prefix: &str, relayer_id: &str) -> Result<()> { - use chrono::Utc; - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}"); - let field = format!("{relayer_id}:last_sync"); - let timestamp = Utc::now().timestamp(); - - conn.hset::<_, _, _, ()>(&hash_key, &field, timestamp) - .await - .map_err(|e| eyre::eyre!("Failed to set relayer last sync: {}", e))?; - - debug!( - relayer_id = %relayer_id, - timestamp = %timestamp, - "recorded relayer last sync time" - ); - - Ok(()) -} - -/// Gets the last sync timestamp for a relayer. -/// -/// # Arguments -/// * `conn` - Redis connection manager -/// * `prefix` - Key prefix for multi-tenant support -/// * `relayer_id` - The relayer's unique identifier -/// -/// # Returns -/// * `Ok(Some(DateTime))` - The last sync time if recorded -/// * `Ok(None)` - If the relayer has never been synced -/// * `Err(_)` - Redis communication error -pub async fn get_relayer_last_sync( - pool: &Arc, - prefix: &str, - relayer_id: &str, -) -> Result>> { - use chrono::{TimeZone, Utc}; - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}"); - let field = format!("{relayer_id}:last_sync"); - - let timestamp: Option = conn - .hget(&hash_key, &field) - .await - .map_err(|e| eyre::eyre!("Failed to get relayer last sync: {}", e))?; - - Ok(timestamp.and_then(|ts| Utc.timestamp_opt(ts, 0).single())) -} - -/// Checks if a relayer was recently synced within the given threshold. -/// -/// This is used to skip initialization for relayers that were recently -/// initialized by another instance (e.g., during rolling restarts). -/// -/// # Arguments -/// * `conn` - Redis connection manager -/// * `prefix` - Key prefix for multi-tenant support -/// * `relayer_id` - The relayer's unique identifier -/// * `threshold_secs` - Number of seconds to consider as "recent" -/// -/// # Returns -/// * `Ok(true)` - If the relayer was synced within the threshold -/// * `Ok(false)` - If the relayer was not synced or sync is stale -/// * `Err(_)` - Redis communication error -pub async fn is_relayer_recently_synced( - pool: &Arc, - prefix: &str, - relayer_id: &str, - threshold_secs: u64, -) -> Result { - use chrono::Utc; - - let last_sync = get_relayer_last_sync(pool, prefix, relayer_id).await?; - - match last_sync { - Some(sync_time) => { - let elapsed = Utc::now().signed_duration_since(sync_time); - let is_recent = elapsed.num_seconds() < threshold_secs as i64; - - if is_recent { - debug!( - relayer_id = %relayer_id, - last_sync = %sync_time.to_rfc3339(), - elapsed_secs = %elapsed.num_seconds(), - threshold_secs = %threshold_secs, - "relayer was recently synced" - ); - } - - Ok(is_recent) - } - None => Ok(false), - } -} - -// ============================================================================ -// Global Initialization Tracking Functions -// ============================================================================ -// -// These functions track when the global relayer initialization was last completed. -// This allows multiple instances to skip redundant initialization when -// initialization was recently completed by another instance. - -/// Sets the global initialization completion timestamp to the current time. -/// -/// This should be called after all relayers have been successfully initialized -/// to record when the initialization occurred. -/// -/// # Arguments -/// * `conn` - Redis connection manager -/// * `prefix` - Key prefix for multi-tenant support -/// -/// # Redis Key Format -/// Hash key: `{prefix}:relayer_sync_meta` -/// Hash field: `global:init_completed` -/// Value: Unix timestamp in seconds -pub async fn set_global_init_completed(pool: &Arc, prefix: &str) -> Result<()> { - use chrono::Utc; - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}"); - let timestamp = Utc::now().timestamp(); - - conn.hset::<_, _, _, ()>(&hash_key, GLOBAL_INIT_FIELD, timestamp) - .await - .map_err(|e| eyre::eyre!("Failed to set global init completed: {}", e))?; - - debug!(timestamp = %timestamp, "recorded global initialization completion time"); - Ok(()) -} - -/// Checks if global initialization was recently completed within the given threshold. -/// -/// This is used to skip initialization when another instance recently completed -/// initialization (e.g., during rolling restarts). -/// -/// # Arguments -/// * `conn` - Redis connection manager -/// * `prefix` - Key prefix for multi-tenant support -/// * `threshold_secs` - Number of seconds to consider as "recent" -/// -/// # Returns -/// * `Ok(true)` - If initialization was completed within the threshold -/// * `Ok(false)` - If initialization was not completed or is stale -/// * `Err(_)` - Redis communication error -pub async fn is_global_init_recently_completed( - pool: &Arc, - prefix: &str, - threshold_secs: u64, -) -> Result { - use chrono::Utc; - use redis::AsyncCommands; - - let mut conn = pool.get().await?; - let hash_key = format!("{prefix}:{RELAYER_SYNC_META_KEY}"); - - let timestamp: Option = conn - .hget(&hash_key, GLOBAL_INIT_FIELD) - .await - .map_err(|e| eyre::eyre!("Failed to get global init time: {}", e))?; - - match timestamp { - Some(ts) => { - let now = Utc::now().timestamp(); - let elapsed = now - ts; - let is_recent = elapsed < threshold_secs as i64; - - if is_recent { - debug!( - elapsed_secs = %elapsed, - threshold_secs = %threshold_secs, - "global initialization recently completed" - ); - } - - Ok(is_recent) - } - None => Ok(false), - } -} - #[cfg(test)] mod tests { use super::*; @@ -1175,629 +846,5 @@ mod tests { // All should work without issues (backward compatible) } - // ===================================================================== - // Relayer Sync Metadata Tests - // ===================================================================== - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_set_and_get_relayer_last_sync() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_sync"; - let relayer_id = "test-relayer-sync"; - - // Set the last sync time - set_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should set last sync time"); - - // Get the last sync time - let last_sync = get_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should get last sync time"); - - assert!(last_sync.is_some(), "Should have a last sync time"); - - // Verify the timestamp is recent (within last minute) - let elapsed = chrono::Utc::now() - .signed_duration_since(last_sync.unwrap()) - .num_seconds(); - assert!(elapsed < 60, "Last sync should be within last minute"); - - // Cleanup: delete the hash - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_get_relayer_last_sync_returns_none_when_not_set() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_sync_none"; - let relayer_id = "nonexistent-relayer"; - - // Get the last sync time for a relayer that hasn't been synced - let last_sync = get_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should not error"); - - assert!( - last_sync.is_none(), - "Should return None for unsynced relayer" - ); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_returns_true_for_recent_sync() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_recent_sync"; - let relayer_id = "recent-relayer"; - - // Set the last sync time - set_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should set last sync time"); - - // Check if recently synced (within 5 minutes) - let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300) - .await - .expect("Should check recent sync"); - - assert!(is_recent, "Should be recently synced"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_returns_false_when_not_set() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_not_synced"; - let relayer_id = "never-synced-relayer"; - - // Check if recently synced for a relayer that hasn't been synced - let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300) - .await - .expect("Should check recent sync"); - - assert!(!is_recent, "Should not be recently synced"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_returns_false_for_stale_sync() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_stale_sync"; - let relayer_id = "stale-relayer"; - - // Manually set an old timestamp (10 minutes ago) - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let field = format!("{relayer_id}:last_sync"); - let old_timestamp = chrono::Utc::now().timestamp() - 600; // 10 minutes ago - - redis::AsyncCommands::hset::<_, _, _, ()>( - &mut conn_clone, - &hash_key, - &field, - old_timestamp, - ) - .await - .expect("Should set old timestamp"); - - // Check if recently synced with 5 minute threshold - let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300) - .await - .expect("Should check recent sync"); - - assert!(!is_recent, "Should not be recently synced (stale)"); - - // Cleanup - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_get_relayer_last_sync_multiple_relayers() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_multi_sync"; - - // Set sync times for multiple relayers - set_relayer_last_sync(&conn, prefix, "relayer-1") - .await - .expect("Should set sync time"); - tokio::time::sleep(Duration::from_millis(10)).await; - set_relayer_last_sync(&conn, prefix, "relayer-2") - .await - .expect("Should set sync time"); - - let sync1 = get_relayer_last_sync(&conn, prefix, "relayer-1") - .await - .expect("Should get sync time"); - let sync2 = get_relayer_last_sync(&conn, prefix, "relayer-2") - .await - .expect("Should get sync time"); - - assert!(sync1.is_some(), "Relayer 1 should have sync time"); - assert!(sync2.is_some(), "Relayer 2 should have sync time"); - assert!( - sync2.unwrap() >= sync1.unwrap(), - "Relayer 2 should be synced at same time or after relayer 1" - ); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_get_relayer_last_sync_update_existing() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_update_sync"; - let relayer_id = "update-relayer"; - - // Set initial sync time - set_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should set sync time"); - - let first_sync = get_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should get sync time") - .expect("Should have sync time"); - - // Wait and update - tokio::time::sleep(Duration::from_millis(100)).await; - - set_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should update sync time"); - - let second_sync = get_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should get sync time") - .expect("Should have sync time"); - - assert!( - second_sync > first_sync, - "Updated sync time should be later than first" - ); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_threshold_boundary() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_boundary_sync"; - let relayer_id = "boundary-relayer"; - - // Set timestamp exactly at threshold (should be considered NOT recent) - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let field = format!("{relayer_id}:last_sync"); - let threshold_secs: u64 = 60; - let boundary_timestamp = chrono::Utc::now().timestamp() - (threshold_secs as i64); - - redis::AsyncCommands::hset::<_, _, _, ()>( - &mut conn_clone, - &hash_key, - &field, - boundary_timestamp, - ) - .await - .expect("Should set boundary timestamp"); - - let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, threshold_secs) - .await - .expect("Should check recent sync"); - - assert!(!is_recent, "Should not be recent at exact threshold"); - - // Cleanup - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_just_before_threshold() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_just_before_sync"; - let relayer_id = "just-before-relayer"; - - // Set timestamp just before threshold (should be considered recent) - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let field = format!("{relayer_id}:last_sync"); - let threshold_secs: u64 = 60; - let just_before_timestamp = - chrono::Utc::now().timestamp() - (threshold_secs as i64) + 5; - - redis::AsyncCommands::hset::<_, _, _, ()>( - &mut conn_clone, - &hash_key, - &field, - just_before_timestamp, - ) - .await - .expect("Should set just-before timestamp"); - - let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, threshold_secs) - .await - .expect("Should check recent sync"); - - assert!(is_recent, "Should be recent just before threshold"); - - // Cleanup - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_different_prefixes() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let relayer_id = "shared-relayer"; - - // Set sync for prefix1 only - set_relayer_last_sync(&conn, "prefix1", relayer_id) - .await - .expect("Should set sync time"); - - let is_recent_prefix1 = is_relayer_recently_synced(&conn, "prefix1", relayer_id, 300) - .await - .expect("Should check sync"); - let is_recent_prefix2 = is_relayer_recently_synced(&conn, "prefix2", relayer_id, 300) - .await - .expect("Should check sync"); - - assert!(is_recent_prefix1, "Should be recent for prefix1"); - assert!(!is_recent_prefix2, "Should not be recent for prefix2"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = "prefix1:relayer_sync_meta"; - let _: () = redis::AsyncCommands::del(&mut conn_clone, hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_relayer_recently_synced_zero_threshold() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_zero_threshold"; - let relayer_id = "zero-threshold-relayer"; - - set_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should set sync time"); - - // With zero threshold, even immediate sync should be considered stale - let is_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 0) - .await - .expect("Should check sync"); - - assert!(!is_recent, "Should not be recent with zero threshold"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - // ===================================================================== - // Global Initialization Tracking Tests - // ===================================================================== - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_set_and_check_global_init_completed() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_global_init"; - - // Set global init completed - set_global_init_completed(&conn, prefix) - .await - .expect("Should set global init completed"); - - // Check if recently completed (within 5 minutes) - let is_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check global init"); - - assert!(is_recent, "Should be recently completed"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_global_init_recently_completed_returns_false_when_not_set() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_global_init_not_set"; - - // Check without setting - should return false - let is_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check global init"); - - assert!(!is_recent, "Should not be recently completed when not set"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_is_global_init_recently_completed_returns_false_when_stale() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_global_init_stale"; - - // Manually set an old timestamp (10 minutes ago) - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let old_timestamp = chrono::Utc::now().timestamp() - 600; // 10 minutes ago - - redis::AsyncCommands::hset::<_, _, _, ()>( - &mut conn_clone, - &hash_key, - "global:init_completed", - old_timestamp, - ) - .await - .expect("Should set old timestamp"); - - // Check with 5 minute threshold - should be stale - let is_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check global init"); - - assert!(!is_recent, "Should not be recent when stale"); - - // Cleanup - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_global_init_different_prefixes() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - // Set global init for prefix1 only - set_global_init_completed(&conn, "global_prefix1") - .await - .expect("Should set global init"); - - let is_recent_prefix1 = is_global_init_recently_completed(&conn, "global_prefix1", 300) - .await - .expect("Should check global init"); - let is_recent_prefix2 = is_global_init_recently_completed(&conn, "global_prefix2", 300) - .await - .expect("Should check global init"); - - assert!(is_recent_prefix1, "Should be recent for prefix1"); - assert!(!is_recent_prefix2, "Should not be recent for prefix2"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = "global_prefix1:relayer_sync_meta"; - let _: () = redis::AsyncCommands::del(&mut conn_clone, hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_global_init_update_existing() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_global_init_update"; - - // Set initial timestamp - set_global_init_completed(&conn, prefix) - .await - .expect("Should set global init"); - - // Wait a bit - tokio::time::sleep(Duration::from_millis(100)).await; - - // Update timestamp - set_global_init_completed(&conn, prefix) - .await - .expect("Should update global init"); - - // Should still be recent - let is_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check global init"); - - assert!(is_recent, "Should be recent after update"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_global_init_coexists_with_relayer_sync() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_coexist"; - let relayer_id = "test-relayer"; - - // Set both global init and relayer sync - set_global_init_completed(&conn, prefix) - .await - .expect("Should set global init"); - set_relayer_last_sync(&conn, prefix, relayer_id) - .await - .expect("Should set relayer sync"); - - // Both should be checkable - let global_recent = is_global_init_recently_completed(&conn, prefix, 300) - .await - .expect("Should check global init"); - let relayer_recent = is_relayer_recently_synced(&conn, prefix, relayer_id, 300) - .await - .expect("Should check relayer sync"); - - assert!(global_recent, "Global init should be recent"); - assert!(relayer_recent, "Relayer sync should be recent"); - - // Cleanup - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:relayer_sync_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - // ===================================================================== - // Config Processing Marker Tests - // ===================================================================== - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_config_processing_in_progress_is_not_completed() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_config_processing_in_progress"; - - set_config_processing_in_progress(&conn, prefix) - .await - .expect("Should set in-progress marker"); - - let completed = is_config_processing_completed(&conn, prefix) - .await - .expect("Should read config processing status"); - - assert!( - !completed, - "In-progress config processing must not be treated as completed" - ); - - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:bootstrap_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } - - #[tokio::test] - #[ignore] // Requires running Redis instance - async fn test_config_processing_completed_requires_explicit_completion_marker() { - let conn = create_test_redis_connection() - .await - .expect("Redis connection required for this test"); - - let prefix = "test_config_processing_completed"; - - set_config_processing_in_progress(&conn, prefix) - .await - .expect("Should set in-progress marker"); - set_config_processing_completed(&conn, prefix) - .await - .expect("Should set completed marker"); - - let completed = is_config_processing_completed(&conn, prefix) - .await - .expect("Should read config processing status"); - - assert!( - completed, - "Completed config processing must require the explicit completion marker" - ); - - let mut conn_clone = conn.get().await.expect("Failed to get connection"); - let hash_key = format!("{prefix}:bootstrap_meta"); - let _: () = redis::AsyncCommands::del(&mut conn_clone, &hash_key) - .await - .expect("Cleanup failed"); - } } }