use crate::error::{LimiteronError, StorageError};
use ahash::AHashMap as HashMap;
use oxcache::Cache;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
pub type IslandModeCallback = Box<dyn Fn(bool) + Send + Sync>;
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum FallbackStrategy {
FailOpen,
FailClosed,
Degraded,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum ComponentType {
Redis,
Postgres,
L2Cache,
Config,
Ban,
Quota,
Other(String),
}
impl From<&str> for ComponentType {
fn from(s: &str) -> Self {
match s.to_lowercase().as_str() {
"redis" => ComponentType::Redis,
"postgres" => ComponentType::Postgres,
"l2_cache" => ComponentType::L2Cache,
"config" => ComponentType::Config,
"ban" => ComponentType::Ban,
"quota" => ComponentType::Quota,
other => ComponentType::Other(other.to_string()),
}
}
}
impl ComponentType {
pub fn as_str(&self) -> &str {
match self {
ComponentType::Redis => "redis",
ComponentType::Postgres => "postgres",
ComponentType::L2Cache => "l2_cache",
ComponentType::Config => "config",
ComponentType::Ban => "ban",
ComponentType::Quota => "quota",
ComponentType::Other(s) => s,
}
}
}
#[derive(Debug, Clone)]
pub struct FallbackConfig {
pub component: ComponentType,
pub strategy: FallbackStrategy,
pub enabled: bool,
pub timeout: Duration,
pub max_retries: u32,
}
impl Default for FallbackConfig {
fn default() -> Self {
Self {
component: ComponentType::Other("default".to_string()),
strategy: FallbackStrategy::Degraded,
enabled: true,
timeout: Duration::from_secs(5),
max_retries: 3,
}
}
}
impl FallbackConfig {
pub fn new(component: ComponentType, strategy: FallbackStrategy) -> Self {
Self {
component,
strategy,
..Default::default()
}
}
pub fn enabled(mut self, enabled: bool) -> Self {
self.enabled = enabled;
self
}
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self
}
pub fn max_retries(mut self, max_retries: u32) -> Self {
self.max_retries = max_retries;
self
}
}
pub struct FallbackManager {
strategies: Arc<RwLock<HashMap<ComponentType, FallbackConfig>>>,
failure_states: Arc<RwLock<HashMap<ComponentType, bool>>>,
l2_cache: Arc<Cache<String, String>>,
island_mode_callbacks: Arc<RwLock<Vec<IslandModeCallback>>>,
}
impl FallbackManager {
pub fn new(l2_cache: Arc<Cache<String, String>>) -> Self {
log::info!("创建降级策略管理器");
let mut strategies = HashMap::new();
strategies.insert(
ComponentType::Redis,
FallbackConfig::new(ComponentType::Redis, FallbackStrategy::Degraded),
);
strategies.insert(
ComponentType::Postgres,
FallbackConfig::new(ComponentType::Postgres, FallbackStrategy::Degraded),
);
strategies.insert(
ComponentType::L2Cache,
FallbackConfig::new(ComponentType::L2Cache, FallbackStrategy::Degraded),
);
strategies.insert(
ComponentType::Config,
FallbackConfig::new(ComponentType::Config, FallbackStrategy::FailClosed),
);
strategies.insert(
ComponentType::Ban,
FallbackConfig::new(ComponentType::Ban, FallbackStrategy::Degraded),
);
strategies.insert(
ComponentType::Quota,
FallbackConfig::new(ComponentType::Quota, FallbackStrategy::Degraded),
);
Self {
strategies: Arc::new(RwLock::new(strategies)),
failure_states: Arc::new(RwLock::new(HashMap::new())),
l2_cache,
island_mode_callbacks: Arc::new(RwLock::new(Vec::new())),
}
}
pub fn l2_cache(&self) -> &Arc<Cache<String, String>> {
&self.l2_cache
}
pub async fn set_strategy(&self, component: ComponentType, config: FallbackConfig) {
log::info!(
target: "fallback",
"设置降级策略: component={:?}, strategy={:?}",
component, config.strategy
);
let mut strategies = self.strategies.write().await;
strategies.insert(component, config);
}
pub async fn get_strategy(&self, component: ComponentType) -> Option<FallbackConfig> {
let strategies = self.strategies.read().await;
strategies.get(&component).cloned()
}
pub async fn execute_with_fallback<F, Fut, FB, FBFut, T>(
&self,
component: ComponentType,
operation: F,
fallback_operation: FB,
) -> Result<T, LimiteronError>
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<T, LimiteronError>>,
FB: FnOnce() -> FBFut,
FBFut: std::future::Future<Output = Result<T, LimiteronError>>,
{
let config = self
.get_strategy(component.clone())
.await
.unwrap_or_default();
if !config.enabled {
return operation().await;
}
let _is_failed = {
let states = self.failure_states.read().await;
*states.get(&component).unwrap_or(&false)
};
let result = operation().await;
match result {
Ok(value) => {
self.clear_failure_internal(component).await;
Ok(value)
}
Err(e) => {
log::warn!(target: "fallback", "组件操作失败: component={:?}, error={}", component, e);
self.set_failure_internal(component.clone()).await;
self.execute_fallback(component, config, fallback_operation)
.await
}
}
}
async fn execute_fallback<F, Fut, T>(
&self,
component: ComponentType,
config: FallbackConfig,
fallback_operation: F,
) -> Result<T, LimiteronError>
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<T, LimiteronError>>,
{
log::info!(
target: "fallback",
"执行降级策略: component={:?}, strategy={:?}",
component, config.strategy
);
match config.strategy {
FallbackStrategy::FailOpen => {
log::warn!(target: "fallback", "降级策略: FailOpen - 允许请求通过");
Err(LimiteronError::LimitError(
"服务降级,但允许请求通过".to_string(),
))
}
FallbackStrategy::FailClosed => {
log::error!(target: "fallback", "降级策略: FailClosed - 拒绝请求");
Err(LimiteronError::StorageError(StorageError::ConnectionError(
"服务降级,拒绝请求".to_string(),
)))
}
FallbackStrategy::Degraded => {
log::debug!(target: "fallback", "降级策略: Degraded - 使用备用方案");
fallback_operation().await
}
}
}
async fn set_failure_internal(&self, component: ComponentType) {
log::warn!(target: "fallback", "组件故障: {:?}", component);
let mut states = self.failure_states.write().await;
states.insert(component, true);
}
async fn clear_failure_internal(&self, component: ComponentType) {
let mut states = self.failure_states.write().await;
states.remove(&component);
log::info!(target: "fallback", "组件恢复: {:?}", component);
}
pub async fn record_failure(&self, component: ComponentType, _error: &str) {
log::warn!(target: "fallback", "组件故障记录: {:?}", component);
self.set_failure_internal(component).await;
}
pub async fn get_failure_count(&self, component: ComponentType) -> u32 {
let states = self.failure_states.read().await;
if *states.get(&component).unwrap_or(&false) {
1
} else {
0
}
}
pub async fn is_failed(&self, component: ComponentType) -> bool {
let states = self.failure_states.read().await;
*states.get(&component).unwrap_or(&false)
}
pub async fn inject_failure(&self, component: ComponentType) {
log::warn!(target: "fallback", "注入故障: {:?}", component);
self.set_failure_internal(component).await;
}
pub async fn recover_failure(&self, component: ComponentType) {
log::info!(target: "fallback", "恢复故障: {:?}", component);
self.clear_failure_internal(component).await;
}
pub async fn get_all_failures(&self) -> Vec<ComponentType> {
let states = self.failure_states.read().await;
states
.iter()
.filter(|&(_, &failed)| failed)
.map(|(component, _)| component.clone())
.collect()
}
pub async fn register_island_mode_callback(&self, callback: IslandModeCallback) {
let mut callbacks = self.island_mode_callbacks.write().await;
callbacks.push(callback);
log::info!(target: "fallback", "注册孤岛模式通知回调");
}
async fn notify_island_mode_change(&self, is_island: bool) {
let callbacks = self.island_mode_callbacks.read().await;
for callback in callbacks.iter() {
callback(is_island);
}
if is_island {
log::warn!(target: "fallback", "已通知所有回调:进入孤岛模式");
} else {
log::info!(target: "fallback", "已通知所有回调:退出孤岛模式");
}
}
pub async fn set_failure(&self, component: ComponentType) {
log::warn!(target: "fallback", "组件故障: {:?}", component);
let mut states = self.failure_states.write().await;
let was_failed = states.values().any(|&f| f);
states.insert(component.clone(), true);
if !was_failed {
log::error!(target: "fallback", "存储层首次故障,触发孤岛模式");
self.notify_island_mode_change(true).await;
}
}
pub async fn clear_failure(&self, component: ComponentType) {
let mut states = self.failure_states.write().await;
states.remove(&component);
log::info!(target: "fallback", "组件恢复: {:?}", component);
let still_failed = states.values().any(|&f| f);
if !still_failed {
log::info!(target: "fallback", "所有存储层恢复,退出孤岛模式");
drop(states);
self.notify_island_mode_change(false).await;
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_fallback_strategy_display() {
let strategy = FallbackStrategy::FailOpen;
assert_eq!(format!("{:?}", strategy), "FailOpen");
let strategy = FallbackStrategy::FailClosed;
assert_eq!(format!("{:?}", strategy), "FailClosed");
let strategy = FallbackStrategy::Degraded;
assert_eq!(format!("{:?}", strategy), "Degraded");
}
#[test]
fn test_component_type_from_str() {
assert_eq!(ComponentType::from("redis"), ComponentType::Redis);
assert_eq!(ComponentType::from("postgres"), ComponentType::Postgres);
assert_eq!(ComponentType::from("l2_cache"), ComponentType::L2Cache);
assert_eq!(
ComponentType::from("other"),
ComponentType::Other("other".to_string())
);
}
#[test]
fn test_component_type_as_str() {
assert_eq!(ComponentType::Redis.as_str(), "redis");
assert_eq!(ComponentType::Postgres.as_str(), "postgres");
assert_eq!(ComponentType::L2Cache.as_str(), "l2_cache");
}
#[test]
fn test_fallback_config_default() {
let config = FallbackConfig::default();
assert_eq!(config.strategy, FallbackStrategy::Degraded);
assert!(config.enabled);
assert_eq!(config.timeout, Duration::from_secs(5));
assert_eq!(config.max_retries, 3);
}
#[test]
fn test_fallback_config_builder() {
let config = FallbackConfig::new(ComponentType::Redis, FallbackStrategy::FailOpen)
.enabled(false)
.timeout(Duration::from_secs(10))
.max_retries(5);
assert_eq!(config.component, ComponentType::Redis);
assert_eq!(config.strategy, FallbackStrategy::FailOpen);
assert!(!config.enabled);
assert_eq!(config.timeout, Duration::from_secs(10));
assert_eq!(config.max_retries, 5);
}
#[tokio::test]
async fn test_fallback_manager_new() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let redis_strategy = manager.get_strategy(ComponentType::Redis).await;
assert!(redis_strategy.is_some());
assert_eq!(redis_strategy.unwrap().strategy, FallbackStrategy::Degraded);
let postgres_strategy = manager.get_strategy(ComponentType::Postgres).await;
assert!(postgres_strategy.is_some());
assert_eq!(
postgres_strategy.unwrap().strategy,
FallbackStrategy::Degraded
);
}
#[tokio::test]
async fn test_fallback_manager_set_strategy() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let config = FallbackConfig::new(ComponentType::Redis, FallbackStrategy::FailOpen);
manager.set_strategy(ComponentType::Redis, config).await;
let strategy = manager.get_strategy(ComponentType::Redis).await;
assert!(strategy.is_some());
assert_eq!(strategy.unwrap().strategy, FallbackStrategy::FailOpen);
}
#[tokio::test]
async fn test_fallback_manager_execute_success() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let result = manager
.execute_with_fallback(
ComponentType::Redis,
|| async { Ok::<String, LimiteronError>("primary".to_string()) },
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), "primary");
assert!(!manager.is_failed(ComponentType::Redis).await);
}
#[tokio::test]
async fn test_fallback_manager_execute_fail_degraded() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let config = FallbackConfig::new(ComponentType::Redis, FallbackStrategy::Degraded);
manager.set_strategy(ComponentType::Redis, config).await;
let result = manager
.execute_with_fallback(
ComponentType::Redis,
|| async {
Err::<String, LimiteronError>(LimiteronError::StorageError(
StorageError::ConnectionError("connection failed".to_string()),
))
},
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), "fallback");
assert!(manager.is_failed(ComponentType::Redis).await);
}
#[tokio::test]
async fn test_fallback_manager_execute_fail_fail_open() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let config = FallbackConfig::new(ComponentType::Redis, FallbackStrategy::FailOpen);
manager.set_strategy(ComponentType::Redis, config).await;
let result = manager
.execute_with_fallback(
ComponentType::Redis,
|| async {
Err::<String, LimiteronError>(LimiteronError::StorageError(
StorageError::ConnectionError("connection failed".to_string()),
))
},
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_err());
assert!(
result
.unwrap_err()
.to_string()
.contains("服务降级,但允许请求通过")
);
}
#[tokio::test]
async fn test_fallback_manager_execute_fail_fail_closed() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let config = FallbackConfig::new(ComponentType::Redis, FallbackStrategy::FailClosed);
manager.set_strategy(ComponentType::Redis, config).await;
let result = manager
.execute_with_fallback(
ComponentType::Redis,
|| async {
Err::<String, LimiteronError>(LimiteronError::StorageError(
StorageError::ConnectionError("connection failed".to_string()),
))
},
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_err());
assert!(
result
.unwrap_err()
.to_string()
.contains("服务降级,拒绝请求")
);
}
#[tokio::test]
async fn test_fallback_manager_inject_failure() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
assert!(!manager.is_failed(ComponentType::Redis).await);
manager.inject_failure(ComponentType::Redis).await;
assert!(manager.is_failed(ComponentType::Redis).await);
manager.recover_failure(ComponentType::Redis).await;
assert!(!manager.is_failed(ComponentType::Redis).await);
}
#[tokio::test]
async fn test_fallback_manager_get_all_failures() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
manager.inject_failure(ComponentType::Redis).await;
manager.inject_failure(ComponentType::Postgres).await;
let failures = manager.get_all_failures().await;
assert_eq!(failures.len(), 2);
assert!(failures.contains(&ComponentType::Redis));
assert!(failures.contains(&ComponentType::Postgres));
}
#[tokio::test]
async fn test_fallback_manager_recovery() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let result = manager
.execute_with_fallback(
ComponentType::Redis,
|| async {
Err::<String, LimiteronError>(LimiteronError::StorageError(
StorageError::ConnectionError("connection failed".to_string()),
))
},
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_ok());
assert!(manager.is_failed(ComponentType::Redis).await);
let result = manager
.execute_with_fallback(
ComponentType::Redis,
|| async { Ok::<String, LimiteronError>("recovered".to_string()) },
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), "recovered");
assert!(!manager.is_failed(ComponentType::Redis).await);
}
#[tokio::test]
async fn test_fallback_manager_l2_cache() {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
let l2_cache = Arc::new(cache);
let manager = FallbackManager::new(l2_cache);
let cache = manager.l2_cache();
assert_eq!(cache.len().await.unwrap(), 0);
}
#[test]
fn test_component_type_from_str_all_variants() {
assert_eq!(ComponentType::from("l2_cache"), ComponentType::L2Cache);
assert_eq!(ComponentType::from("config"), ComponentType::Config);
assert_eq!(ComponentType::from("ban"), ComponentType::Ban);
assert_eq!(ComponentType::from("quota"), ComponentType::Quota);
assert_eq!(
ComponentType::from("custom_service"),
ComponentType::Other("custom_service".to_string())
);
}
#[test]
fn test_component_type_from_str_case_insensitive() {
assert_eq!(ComponentType::from("REDIS"), ComponentType::Redis);
assert_eq!(ComponentType::from("Postgres"), ComponentType::Postgres);
assert_eq!(ComponentType::from("L2_CACHE"), ComponentType::L2Cache);
assert_eq!(ComponentType::from("Config"), ComponentType::Config);
assert_eq!(ComponentType::from("BAN"), ComponentType::Ban);
assert_eq!(ComponentType::from("QUOTA"), ComponentType::Quota);
}
#[test]
fn test_component_type_as_str_all_variants() {
assert_eq!(ComponentType::Redis.as_str(), "redis");
assert_eq!(ComponentType::Postgres.as_str(), "postgres");
assert_eq!(ComponentType::L2Cache.as_str(), "l2_cache");
assert_eq!(ComponentType::Config.as_str(), "config");
assert_eq!(ComponentType::Ban.as_str(), "ban");
assert_eq!(ComponentType::Quota.as_str(), "quota");
assert_eq!(
ComponentType::Other("custom_type".to_string()).as_str(),
"custom_type"
);
}
#[test]
fn test_component_type_other_equality() {
assert_eq!(
ComponentType::Other("a".to_string()),
ComponentType::Other("a".to_string())
);
assert_ne!(
ComponentType::Other("a".to_string()),
ComponentType::Other("b".to_string())
);
assert_ne!(ComponentType::Other("a".to_string()), ComponentType::Redis);
}
#[test]
fn test_component_type_serde_roundtrip() {
let variants = vec![
ComponentType::Redis,
ComponentType::Postgres,
ComponentType::L2Cache,
ComponentType::Config,
ComponentType::Ban,
ComponentType::Quota,
ComponentType::Other("custom".to_string()),
];
for variant in variants {
let json = serde_json::to_string(&variant).unwrap();
let deserialized: ComponentType = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized, variant);
}
}
#[test]
fn test_fallback_strategy_serde_roundtrip() {
let variants = vec![
FallbackStrategy::FailOpen,
FallbackStrategy::FailClosed,
FallbackStrategy::Degraded,
];
for variant in variants {
let json = serde_json::to_string(&variant).unwrap();
let deserialized: FallbackStrategy = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized, variant);
}
}
#[test]
fn test_fallback_strategy_debug() {
assert_eq!(format!("{:?}", FallbackStrategy::FailOpen), "FailOpen");
assert_eq!(format!("{:?}", FallbackStrategy::FailClosed), "FailClosed");
assert_eq!(format!("{:?}", FallbackStrategy::Degraded), "Degraded");
}
#[test]
fn test_fallback_config_new_defaults() {
let config = FallbackConfig::new(ComponentType::Redis, FallbackStrategy::FailOpen);
assert_eq!(config.component, ComponentType::Redis);
assert_eq!(config.strategy, FallbackStrategy::FailOpen);
assert!(config.enabled);
assert_eq!(config.timeout, Duration::from_secs(5));
assert_eq!(config.max_retries, 3);
}
#[test]
fn test_fallback_config_new_other_component() {
let config = FallbackConfig::new(
ComponentType::Other("custom".to_string()),
FallbackStrategy::FailClosed,
);
assert_eq!(config.component, ComponentType::Other("custom".to_string()));
assert_eq!(config.strategy, FallbackStrategy::FailClosed);
assert!(config.enabled);
assert_eq!(config.max_retries, 3);
}
async fn create_manager() -> FallbackManager {
let cache: Cache<String, String> = Cache::builder()
.capacity(10000)
.ttl(Duration::from_secs(60))
.build()
.await
.unwrap();
FallbackManager::new(Arc::new(cache))
}
#[tokio::test]
async fn test_fallback_manager_all_default_strategies() {
let manager = create_manager().await;
let redis = manager.get_strategy(ComponentType::Redis).await.unwrap();
assert_eq!(redis.strategy, FallbackStrategy::Degraded);
let postgres = manager.get_strategy(ComponentType::Postgres).await.unwrap();
assert_eq!(postgres.strategy, FallbackStrategy::Degraded);
let l2 = manager.get_strategy(ComponentType::L2Cache).await.unwrap();
assert_eq!(l2.strategy, FallbackStrategy::Degraded);
let config = manager.get_strategy(ComponentType::Config).await.unwrap();
assert_eq!(config.strategy, FallbackStrategy::FailClosed);
let ban = manager.get_strategy(ComponentType::Ban).await.unwrap();
assert_eq!(ban.strategy, FallbackStrategy::Degraded);
let quota = manager.get_strategy(ComponentType::Quota).await.unwrap();
assert_eq!(quota.strategy, FallbackStrategy::Degraded);
}
#[tokio::test]
async fn test_fallback_manager_get_strategy_unknown() {
let manager = create_manager().await;
let result = manager
.get_strategy(ComponentType::Other("nonexistent".to_string()))
.await;
assert!(result.is_none());
}
#[tokio::test]
async fn test_fallback_manager_execute_disabled() {
let manager = create_manager().await;
let config =
FallbackConfig::new(ComponentType::Redis, FallbackStrategy::Degraded).enabled(false);
manager.set_strategy(ComponentType::Redis, config).await;
let result: Result<String, LimiteronError> = manager
.execute_with_fallback(
ComponentType::Redis,
|| async {
Err::<String, LimiteronError>(LimiteronError::LimitError(
"primary failed".to_string(),
))
},
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("primary failed"));
}
#[tokio::test]
async fn test_fallback_manager_execute_disabled_success() {
let manager = create_manager().await;
let config =
FallbackConfig::new(ComponentType::Redis, FallbackStrategy::Degraded).enabled(false);
manager.set_strategy(ComponentType::Redis, config).await;
let result: Result<String, LimiteronError> = manager
.execute_with_fallback(
ComponentType::Redis,
|| async { Ok::<String, LimiteronError>("primary_ok".to_string()) },
|| async { Ok::<String, LimiteronError>("fallback".to_string()) },
)
.await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), "primary_ok");
}
#[tokio::test]
async fn test_fallback_manager_execute_unknown_component() {
let manager = create_manager().await;
let result: Result<String, LimiteronError> = manager
.execute_with_fallback(
ComponentType::Other("unknown".to_string()),
|| async {
Err::<String, LimiteronError>(LimiteronError::StorageError(
StorageError::ConnectionError("down".to_string()),
))
},
|| async { Ok::<String, LimiteronError>("degraded_ok".to_string()) },
)
.await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), "degraded_ok");
}
#[tokio::test]
async fn test_fallback_manager_record_failure() {
let manager = create_manager().await;
assert!(!manager.is_failed(ComponentType::Redis).await);
manager
.record_failure(ComponentType::Redis, "connection timeout")
.await;
assert!(manager.is_failed(ComponentType::Redis).await);
}
#[tokio::test]
async fn test_fallback_manager_failure_count_zero() {
let manager = create_manager().await;
let count = manager.get_failure_count(ComponentType::Redis).await;
assert_eq!(count, 0);
}
#[tokio::test]
async fn test_fallback_manager_failure_count_one() {
let manager = create_manager().await;
manager.inject_failure(ComponentType::Redis).await;
let count = manager.get_failure_count(ComponentType::Redis).await;
assert_eq!(count, 1);
}
#[tokio::test]
async fn test_fallback_manager_set_failure_island_mode_enter() {
let manager = create_manager().await;
let entered = Arc::new(std::sync::Mutex::new(None::<bool>));
let entered_clone = entered.clone();
let callback: IslandModeCallback = Box::new(move |is_island| {
*entered_clone.lock().unwrap() = Some(is_island);
});
manager.register_island_mode_callback(callback).await;
manager.set_failure(ComponentType::Redis).await;
assert!(manager.is_failed(ComponentType::Redis).await);
assert_eq!(*entered.lock().unwrap(), Some(true));
}
#[tokio::test]
async fn test_fallback_manager_clear_failure_island_mode_exit() {
let manager = create_manager().await;
let state = Arc::new(std::sync::Mutex::new(Vec::new()));
let state_clone = state.clone();
let callback: IslandModeCallback = Box::new(move |is_island| {
state_clone.lock().unwrap().push(is_island);
});
manager.register_island_mode_callback(callback).await;
manager.set_failure(ComponentType::Redis).await;
assert!(manager.is_failed(ComponentType::Redis).await);
manager.clear_failure(ComponentType::Redis).await;
assert!(!manager.is_failed(ComponentType::Redis).await);
let calls = state.lock().unwrap();
assert_eq!(calls.len(), 2);
assert!(calls[0]);
assert!(!calls[1]);
}
#[tokio::test]
async fn test_fallback_manager_island_mode_lifecycle() {
let manager = create_manager().await;
let state = Arc::new(std::sync::Mutex::new(Vec::new()));
let state_clone = state.clone();
let callback: IslandModeCallback = Box::new(move |is_island| {
state_clone.lock().unwrap().push(is_island);
});
manager.register_island_mode_callback(callback).await;
manager.set_failure(ComponentType::Redis).await;
{
let calls = state.lock().unwrap();
assert_eq!(calls.len(), 1);
assert!(calls[0]);
}
manager.set_failure(ComponentType::Postgres).await;
{
let calls = state.lock().unwrap();
assert_eq!(calls.len(), 1);
}
manager.clear_failure(ComponentType::Redis).await;
{
let calls = state.lock().unwrap();
assert_eq!(calls.len(), 1);
}
assert!(!manager.is_failed(ComponentType::Redis).await);
assert!(manager.is_failed(ComponentType::Postgres).await);
manager.clear_failure(ComponentType::Postgres).await;
{
let calls = state.lock().unwrap();
assert_eq!(calls.len(), 2);
assert!(!calls[1]);
}
assert!(!manager.is_failed(ComponentType::Postgres).await);
}
#[tokio::test]
async fn test_fallback_manager_set_failure_no_notify_twice() {
let manager = create_manager().await;
let count = Arc::new(std::sync::Mutex::new(0u32));
let count_clone = count.clone();
let callback: IslandModeCallback = Box::new(move |_| {
*count_clone.lock().unwrap() += 1;
});
manager.register_island_mode_callback(callback).await;
manager.set_failure(ComponentType::Redis).await;
manager.set_failure(ComponentType::Postgres).await;
manager.set_failure(ComponentType::Ban).await;
assert_eq!(*count.lock().unwrap(), 1);
}
#[tokio::test]
async fn test_fallback_manager_clear_failure_no_notify_if_still_failed() {
let manager = create_manager().await;
let state = Arc::new(std::sync::Mutex::new(Vec::new()));
let state_clone = state.clone();
let callback: IslandModeCallback = Box::new(move |is_island| {
state_clone.lock().unwrap().push(is_island);
});
manager.register_island_mode_callback(callback).await;
manager.set_failure(ComponentType::Redis).await;
manager.set_failure(ComponentType::Postgres).await;
manager.clear_failure(ComponentType::Redis).await;
let calls = state.lock().unwrap();
assert_eq!(calls.len(), 1);
assert!(calls[0]);
}
#[tokio::test]
async fn test_fallback_manager_clear_failure_no_notify_on_empty() {
let manager = create_manager().await;
let count = Arc::new(std::sync::Mutex::new(0u32));
let count_clone = count.clone();
let callback: IslandModeCallback = Box::new(move |_| {
*count_clone.lock().unwrap() += 1;
});
manager.register_island_mode_callback(callback).await;
manager.clear_failure(ComponentType::Redis).await;
assert_eq!(*count.lock().unwrap(), 1);
}
#[tokio::test]
async fn test_fallback_manager_register_island_mode_callback_multiple() {
let manager = create_manager().await;
let count = Arc::new(std::sync::Mutex::new(0u32));
let c1 = {
let count = count.clone();
Box::new(move |_: bool| {
*count.lock().unwrap() += 1;
}) as IslandModeCallback
};
let c2 = {
let count = count.clone();
Box::new(move |_: bool| {
*count.lock().unwrap() += 1;
}) as IslandModeCallback
};
manager.register_island_mode_callback(c1).await;
manager.register_island_mode_callback(c2).await;
manager.set_failure(ComponentType::Redis).await;
assert_eq!(*count.lock().unwrap(), 2);
}
}