diff --git a/src/bootstrap/config_processor.rs b/src/bootstrap/config_processor.rs index 5b404690c..4237623ef 100644 --- a/src/bootstrap/config_processor.rs +++ b/src/bootstrap/config_processor.rs @@ -1,6 +1,20 @@ //! 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}, @@ -15,10 +29,19 @@ 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; +use tracing::{info, warn}; + +/// Lock name for config processing lock. +const CONFIG_PROCESSING_LOCK_NAME: &str = "config_processing"; async fn process_api_key( server_config: &ServerConfig, @@ -322,6 +345,10 @@ where return Ok(true); } + if app_state.api_key_repository.has_entries().await? { + return Ok(true); + } + Ok(false) } @@ -334,6 +361,10 @@ 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, @@ -350,18 +381,300 @@ where PR: PluginRepositoryTrait + Send + Sync + 'static, AKR: ApiKeyRepositoryTrait + Send + Sync + 'static, { - let should_process_config_file = match server_config.repository_storage_type { - RepositoryStorageType::InMemory => true, + 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 + } RepositoryStorageType::Redis => { - server_config.reset_storage_on_start || !is_redis_populated(app_state).await? + // 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), + ); - if !should_process_config_file { - info!("Skipping config file processing"); - return Ok(()); + 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" + )) + } + } + 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?; + } + + 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?; @@ -373,15 +686,14 @@ where app_state.api_key_repository.drop_all_entries().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?; - } + 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(()) } @@ -1527,6 +1839,18 @@ 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 { @@ -1729,4 +2053,70 @@ 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 059ef56c0..9924ec2ce 100644 --- a/src/bootstrap/initialize_relayers.rs +++ b/src/bootstrap/initialize_relayers.rs @@ -2,7 +2,23 @@ //! //! 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::{ @@ -13,10 +29,23 @@ 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 futures::future::try_join_all; -use tracing::debug; +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"; /// Internal function for initializing a relayer using a provided relayer service. /// This allows for easier testing with mocked relayers. @@ -78,32 +107,339 @@ where // Early return for empty list - no work to do if relayers.is_empty() { - debug!("No relayers to initialize"); + info!("No relayers to initialize"); return Ok(()); } - debug!(count = relayers.len(), "Initializing relayers concurrently"); + 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?; - let relayer_futures = relayers.iter().map(|relayer| { + 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 app_state = app_state.clone(); - async move { initialize_relayer(relayer.id.clone(), app_state).await } + let relayer_id = relayer.id.clone(); + + async move { + let result = initialize_relayer(relayer_id.clone(), app_state).await; + (relayer_id, result) + } }); - try_join_all(relayer_futures) - .await - .wrap_err("Failed to initialize relayers")?; + let results = futures::future::join_all(futures).await; - debug!( - count = relayers.len(), - "All relayers initialized successfully" + // Count and report results + let succeeded = results.iter().filter(|(_, r)| r.is_ok()).count(); + let failed = results.iter().filter(|(_, r)| r.is_err()).count(); + + info!( + succeeded = succeeded, + failed = failed, + "Relayer initialization completed" ); + + // 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::utils::mocks::mockutils::create_mock_relayer; + 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()), + } + } #[test] fn test_get_relayer_ids_with_empty_list() { @@ -299,4 +635,564 @@ 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 561d8f005..a9a3c8e3a 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 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) } @@ -345,18 +345,18 @@ impl ApiKeyRepositoryTrait for RedisApiKeyRepository { 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 = conn - .smembers(&plugin_list_key) + // Get all API key IDs first + let api_key_ids: Vec = 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(()); } @@ -364,20 +364,20 @@ impl ApiKeyRepositoryTrait for RedisApiKeyRepository { 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(()) } } diff --git a/src/repositories/relayer/mod.rs b/src/repositories/relayer/mod.rs index f0f428d60..8067413ec 100644 --- a/src/repositories/relayer/mod.rs +++ b/src/repositories/relayer/mod.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 + 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)] @@ -96,6 +110,7 @@ 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)>; } } @@ -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, 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, 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, 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 b9db30317..0f0981ad2 100644 --- a/src/repositories/relayer/relayer_in_memory.rs +++ b/src/repositories/relayer/relayer_in_memory.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] diff --git a/src/repositories/relayer/relayer_redis.rs b/src/repositories/relayer/relayer_redis.rs index e402e9993..49626ef9e 100644 --- a/src/repositories/relayer/relayer_redis.rs +++ b/src/repositories/relayer/relayer_redis.rs @@ -560,6 +560,10 @@ 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 2f30ace46..0e2ced380 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -8,6 +8,9 @@ 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 new file mode 100644 index 000000000..df07415c2 --- /dev/null +++ b/src/utils/polling.rs @@ -0,0 +1,192 @@ +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 f5e1f4b9f..472033450 100644 --- a/src/utils/redis.rs +++ b/src/utils/redis.rs @@ -7,6 +7,21 @@ 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, @@ -376,6 +391,320 @@ 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::*; @@ -846,5 +1175,629 @@ 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"); + } } }