use crate::error::{Decision, LimiteronError, RejectionMetadata};
use crate::limiters::Limiter;
use ahash::AHashMap;
use log::{info, warn};
use parking_lot::RwLock;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
#[derive(Clone)]
pub struct DecisionNode {
pub id: String,
pub name: String,
pub limiter: Arc<dyn Limiter>,
pub priority: u16,
pub enabled: bool,
pub short_circuit: bool,
pub cost: u64,
}
impl DecisionNode {
pub fn with_dependencies(
id: String,
name: String,
limiter: Arc<dyn Limiter>,
priority: u16,
) -> Self {
Self {
id,
name,
limiter,
priority,
enabled: true,
short_circuit: true,
cost: 1,
}
}
pub fn new(id: String, name: String, limiter: Arc<dyn Limiter>, priority: u16) -> Self {
Self::with_dependencies(id, name, limiter, priority)
}
pub fn with_enabled(mut self, enabled: bool) -> Self {
self.enabled = enabled;
self
}
pub fn with_short_circuit(mut self, short_circuit: bool) -> Self {
self.short_circuit = short_circuit;
self
}
pub fn with_cost(mut self, cost: u64) -> Self {
self.cost = cost;
self
}
}
pub struct AtomicChainStats {
total_checks: AtomicU64,
allowed_count: AtomicU64,
rejected_count: AtomicU64,
error_count: AtomicU64,
node_rejections: RwLock<AHashMap<String, u64>>,
}
impl AtomicChainStats {
pub fn new() -> Self {
Self {
total_checks: AtomicU64::new(0),
allowed_count: AtomicU64::new(0),
rejected_count: AtomicU64::new(0),
error_count: AtomicU64::new(0),
node_rejections: RwLock::new(AHashMap::new()),
}
}
#[inline]
pub fn increment_total(&self) {
self.total_checks.fetch_add(1, Ordering::Relaxed);
}
#[inline]
pub fn increment_allowed(&self) {
self.allowed_count.fetch_add(1, Ordering::Relaxed);
}
#[inline]
pub fn increment_rejected(&self) {
self.rejected_count.fetch_add(1, Ordering::Relaxed);
}
#[inline]
pub fn increment_error(&self) {
self.error_count.fetch_add(1, Ordering::Relaxed);
}
pub fn increment_node_rejection(&self, node_id: &str) {
let mut rejections = self.node_rejections.write();
*rejections.entry(node_id.to_string()).or_insert(0) += 1;
}
pub fn snapshot(&self) -> ChainStats {
let node_rejections: Vec<(String, u64)> = self
.node_rejections
.read()
.iter()
.map(|(k, v)| (k.clone(), *v))
.collect();
ChainStats {
total_checks: self.total_checks.load(Ordering::Relaxed),
allowed_count: self.allowed_count.load(Ordering::Relaxed),
rejected_count: self.rejected_count.load(Ordering::Relaxed),
error_count: self.error_count.load(Ordering::Relaxed),
node_rejections,
}
}
pub fn reset(&self) {
self.total_checks.store(0, Ordering::Relaxed);
self.allowed_count.store(0, Ordering::Relaxed);
self.rejected_count.store(0, Ordering::Relaxed);
self.error_count.store(0, Ordering::Relaxed);
self.node_rejections.write().clear();
}
}
impl Default for AtomicChainStats {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone)]
pub struct DecisionChain {
nodes: Vec<DecisionNode>,
stats: Arc<AtomicChainStats>,
}
#[derive(Debug, Clone, Default)]
pub struct ChainStats {
pub total_checks: u64,
pub allowed_count: u64,
pub rejected_count: u64,
pub error_count: u64,
pub node_rejections: Vec<(String, u64)>,
}
impl DecisionChain {
pub fn builder() -> DecisionChainBuilder {
DecisionChainBuilder::new()
}
pub fn with_dependencies(nodes: Vec<DecisionNode>) -> Self {
let mut chain = Self {
nodes: Vec::new(),
stats: Arc::new(AtomicChainStats::new()),
};
for node in nodes {
chain.add_node(node);
}
chain
}
pub async fn check(&self) -> Result<Decision, LimiteronError> {
let enabled_nodes: Vec<&DecisionNode> = self.nodes.iter().filter(|n| n.enabled).collect();
let mut last_rejection: Option<RejectionMetadata> = None;
for node in enabled_nodes {
match node.limiter.allow(node.cost).await {
Ok(false) => {
self.stats.increment_total();
self.stats.increment_rejected();
self.stats.increment_node_rejection(&node.id);
if node.short_circuit {
return Ok(Decision::Rejected(RejectionMetadata {
reason: format!("Rejected by {}: rate limit exceeded", node.name),
retry_after: 60,
limit: 0,
reset_at: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
+ 60,
}));
}
last_rejection = Some(RejectionMetadata {
reason: format!("Rejected by {}: rate limit exceeded", node.name),
retry_after: 60,
limit: 0,
reset_at: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
+ 60,
});
}
Err(e) => {
self.stats.increment_total();
self.stats.increment_error();
return Err(e);
}
_ => {} }
}
self.stats.increment_total();
if let Some(rejection) = last_rejection {
self.stats.increment_allowed();
return Ok(Decision::Rejected(rejection));
}
self.stats.increment_allowed();
Ok(Decision::allowed_default())
}
pub async fn stats(&self) -> ChainStats {
self.stats.snapshot()
}
pub fn stats_sync(&self) -> ChainStats {
self.stats.snapshot()
}
pub async fn reset_stats(&self) {
self.stats.reset();
}
pub fn reset_stats_sync(&self) {
self.stats.reset();
}
pub fn node_count(&self) -> usize {
self.nodes.len()
}
pub fn enabled_node_count(&self) -> usize {
self.nodes.iter().filter(|n| n.enabled).count()
}
pub fn add_node(&mut self, node: DecisionNode) {
self.nodes.push(node);
}
pub fn enable_node(&mut self, node_id: &str) -> bool {
if let Some(node) = self.nodes.iter_mut().find(|n| n.id == node_id) {
node.enabled = true;
info!("Enabled node: {}", node_id);
true
} else {
warn!("Failed to enable node: {} (not found)", node_id);
false
}
}
pub fn disable_node(&mut self, node_id: &str) -> bool {
if let Some(node) = self.nodes.iter_mut().find(|n| n.id == node_id) {
node.enabled = false;
info!("Disabled node: {}", node_id);
true
} else {
warn!("Failed to disable node: {} (not found)", node_id);
false
}
}
pub fn set_short_circuit(&mut self, node_id: &str, short_circuit: bool) -> bool {
if let Some(node) = self.nodes.iter_mut().find(|n| n.id == node_id) {
node.short_circuit = short_circuit;
info!("Set short_circuit={} for node: {}", short_circuit, node_id);
true
} else {
warn!(
"Failed to set short_circuit for node: {} (not found)",
node_id
);
false
}
}
}
pub struct DecisionChainBuilder {
nodes: Vec<DecisionNode>,
}
impl DecisionChainBuilder {
pub fn new() -> Self {
Self { nodes: Vec::new() }
}
pub fn add_node(mut self, node: DecisionNode) -> Self {
self.nodes.push(node);
self
}
pub fn build(self) -> DecisionChain {
DecisionChain::with_dependencies(self.nodes)
}
}
impl Default for DecisionChainBuilder {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
#[allow(deprecated)]
mod tests {
use super::*;
use crate::limiters::{
ConcurrencyLimiter, FixedWindowLimiter, ShardedSlidingWindowLimiter, TokenBucketLimiter,
};
use async_trait::async_trait;
use std::time::Duration;
struct MockLimiter {
allowed: Arc<std::sync::atomic::AtomicBool>,
}
impl MockLimiter {
fn new(allowed: bool) -> Self {
Self {
allowed: Arc::new(std::sync::atomic::AtomicBool::new(allowed)),
}
}
fn set_allowed(&self, v: bool) {
self.allowed.store(v, std::sync::atomic::Ordering::SeqCst);
}
}
#[async_trait]
impl Limiter for MockLimiter {
async fn allow(&self, _cost: u64) -> Result<bool, LimiteronError> {
let a = self.allowed.load(std::sync::atomic::Ordering::SeqCst);
Ok(a)
}
}
struct SpyLimiter {
calls: Arc<std::sync::atomic::AtomicUsize>,
}
impl SpyLimiter {
fn new() -> Self {
Self {
calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
}
}
}
#[async_trait]
impl Limiter for SpyLimiter {
async fn allow(&self, _cost: u64) -> Result<bool, LimiteronError> {
self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(true)
}
}
#[test]
fn test_decision_node_creation() {
let limiter = Arc::new(TokenBucketLimiter::new(100, 10));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
assert_eq!(node.id, "node1");
assert_eq!(node.name, "Token Bucket");
assert_eq!(node.priority, 100);
assert!(node.enabled);
assert!(node.short_circuit);
assert_eq!(node.cost, 1);
}
#[test]
fn test_decision_node_with_options() {
let limiter = Arc::new(TokenBucketLimiter::new(100, 10));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
)
.with_enabled(false)
.with_short_circuit(false)
.with_cost(5);
assert!(!node.enabled);
assert!(!node.short_circuit);
assert_eq!(node.cost, 5);
}
#[tokio::test]
async fn test_decision_chain_empty() {
let chain = DecisionChain::with_dependencies(vec![]);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
#[tokio::test]
async fn test_decision_chain_single_node() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let chain = DecisionChain::with_dependencies(vec![node]);
for _ in 0..10 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[tokio::test]
async fn test_decision_chain_multiple_nodes() {
let limiter1 = Arc::new(TokenBucketLimiter::new(5, 1));
let limiter2 = Arc::new(ShardedSlidingWindowLimiter::new(Duration::from_secs(1), 10));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter1,
100,
);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Sliding Window".to_string(),
limiter2,
50,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..5 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[tokio::test]
async fn test_decision_chain_priority() {
let limiter1 = Arc::new(TokenBucketLimiter::new(10, 1));
let limiter2 = Arc::new(TokenBucketLimiter::new(5, 1));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"Low Priority".to_string(),
limiter1,
50,
);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"High Priority".to_string(),
limiter2,
100,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..5 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
if let Decision::Rejected(metadata) = decision {
assert!(metadata.reason.contains("High Priority"));
}
}
#[tokio::test]
async fn test_decision_chain_disabled_node() {
let limiter1 = Arc::new(TokenBucketLimiter::new(0, 1));
let limiter2 = Arc::new(TokenBucketLimiter::new(10, 1));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"Disabled Node".to_string(),
limiter1,
100,
)
.with_enabled(false);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Active Node".to_string(),
limiter2,
50,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
#[tokio::test]
async fn test_decision_chain_short_circuit() {
let limiter1 = Arc::new(TokenBucketLimiter::new(5, 1));
let limiter2 = Arc::new(TokenBucketLimiter::new(10, 1));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"First Node".to_string(),
limiter1,
100,
);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Second Node".to_string(),
limiter2,
50,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..5 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[tokio::test]
async fn test_decision_chain_no_short_circuit() {
let limiter1 = Arc::new(TokenBucketLimiter::new(5, 1));
let limiter2 = Arc::new(TokenBucketLimiter::new(3, 1));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"First Node".to_string(),
limiter1,
100,
)
.with_short_circuit(false);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Second Node".to_string(),
limiter2,
50,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..3 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[tokio::test]
async fn test_decision_chain_check() {
let limiter1 = Arc::new(TokenBucketLimiter::new(3, 1));
let limiter2 = Arc::new(TokenBucketLimiter::new(5, 1));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"First Node".to_string(),
limiter1,
100,
);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Second Node".to_string(),
limiter2,
50,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..3 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
if let Decision::Rejected(metadata) = decision {
assert!(metadata.reason.contains("First Node"));
}
}
#[tokio::test]
async fn test_decision_chain_stats() {
let limiter = Arc::new(TokenBucketLimiter::new(5, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let chain = DecisionChain::with_dependencies(vec![node]);
for _ in 0..10 {
chain.check().await.unwrap();
}
let stats = chain.stats().await;
assert_eq!(stats.total_checks, 10);
assert_eq!(stats.allowed_count, 5);
assert_eq!(stats.rejected_count, 5);
}
#[tokio::test]
async fn test_decision_chain_node_rejections() {
let limiter1 = Arc::new(TokenBucketLimiter::new(5, 1));
let limiter2 = Arc::new(TokenBucketLimiter::new(3, 1));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"First Node".to_string(),
limiter1,
100,
)
.with_short_circuit(false);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Second Node".to_string(),
limiter2,
50,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..10 {
chain.check().await.unwrap();
}
let stats = chain.stats().await;
assert_eq!(stats.node_rejections.len(), 2);
let node1_rejections = stats
.node_rejections
.iter()
.find(|(id, _)| id == "node1")
.map(|(_, count)| *count)
.unwrap_or(0);
let node2_rejections = stats
.node_rejections
.iter()
.find(|(id, _)| id == "node2")
.map(|(_, count)| *count)
.unwrap_or(0);
assert!(node1_rejections > 0);
assert!(node2_rejections > 0);
}
#[tokio::test]
#[cfg(feature = "legacy_tests")]
async fn test_decision_chain_add_remove_node_disabled() {
}
#[tokio::test]
async fn test_decision_chain_add_node() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let mut chain = DecisionChain::with_dependencies(vec![]);
assert_eq!(chain.node_count(), 0);
chain.add_node(node);
assert_eq!(chain.node_count(), 1);
}
#[tokio::test]
async fn test_decision_chain_enable_disable_node() {
let limiter = Arc::new(TokenBucketLimiter::new(0, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let mut chain = DecisionChain::with_dependencies(vec![node]);
chain.disable_node("node1");
assert_eq!(chain.enabled_node_count(), 0);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
chain.enable_node("node1");
assert_eq!(chain.enabled_node_count(), 1);
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[tokio::test]
#[cfg(feature = "legacy_tests")] async fn test_decision_chain_set_short_circuit() {
let limiter1 = Arc::new(MockLimiter::new(true));
let limiter2_spy = Arc::new(SpyLimiter::new());
let limiter2 = limiter2_spy.clone();
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"Mock Node".to_string(),
limiter1.clone(),
100,
)
.with_short_circuit(false);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Spy Node".to_string(),
limiter2,
50,
);
let mut chain = DecisionChain::with_dependencies(vec![node1, node2]);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
assert_eq!(
limiter2_spy.calls.load(std::sync::atomic::Ordering::SeqCst),
1
);
limiter1.set_allowed(false);
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
assert_eq!(
limiter2_spy.calls.load(std::sync::atomic::Ordering::SeqCst),
2
);
chain.set_short_circuit("node1", true);
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
assert_eq!(
limiter2_spy.calls.load(std::sync::atomic::Ordering::SeqCst),
2
);
limiter1.set_allowed(true);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
assert_eq!(
limiter2_spy.calls.load(std::sync::atomic::Ordering::SeqCst),
3
); }
#[test]
fn test_decision_chain_builder() {
let limiter1 = Arc::new(TokenBucketLimiter::new(100, 10));
let limiter2 = Arc::new(ShardedSlidingWindowLimiter::new(
Duration::from_secs(1),
100,
));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter1,
100,
);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Sliding Window".to_string(),
limiter2,
50,
);
let chain = DecisionChainBuilder::new()
.add_node(node1)
.add_node(node2)
.build();
assert_eq!(chain.node_count(), 2);
}
#[tokio::test]
async fn test_decision_chain_mixed_limiters() {
let limiter1 = Arc::new(TokenBucketLimiter::new(10, 1));
let limiter2 = Arc::new(ShardedSlidingWindowLimiter::new(Duration::from_secs(1), 5));
let limiter3 = Arc::new(FixedWindowLimiter::new(Duration::from_secs(1), 3));
let limiter4 = Arc::new(ConcurrencyLimiter::new(2));
let node1 = DecisionNode::with_dependencies(
"token_bucket".to_string(),
"Token Bucket".to_string(),
limiter1,
100,
);
let node2 = DecisionNode::with_dependencies(
"sliding_window".to_string(),
"Sliding Window".to_string(),
limiter2,
75,
);
let node3 = DecisionNode::with_dependencies(
"fixed_window".to_string(),
"Fixed Window".to_string(),
limiter3,
50,
);
let node4 = DecisionNode::with_dependencies(
"concurrency".to_string(),
"Concurrency".to_string(),
limiter4,
25,
);
let chain = DecisionChain::with_dependencies(vec![node1, node2, node3, node4]);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
let stats = chain.stats().await;
assert_eq!(stats.total_checks, 1);
assert_eq!(stats.allowed_count, 1);
}
#[tokio::test]
async fn test_decision_chain_cost() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
)
.with_cost(2);
let chain = DecisionChain::with_dependencies(vec![node]);
for _ in 0..5 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[tokio::test]
async fn test_decision_chain_reset_stats() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let chain = DecisionChain::with_dependencies(vec![node]);
for _ in 0..5 {
chain.check().await.unwrap();
}
let stats = chain.stats().await;
assert_eq!(stats.total_checks, 5);
chain.reset_stats().await;
let stats = chain.stats().await;
assert_eq!(stats.total_checks, 0);
}
#[tokio::test]
async fn test_decision_chain_concurrent_checks() {
let limiter = Arc::new(TokenBucketLimiter::new(100, 10));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let chain = Arc::new(DecisionChain::with_dependencies(vec![node]));
let mut handles = vec![];
for _ in 0..10 {
let chain_clone = Arc::clone(&chain);
handles.push(tokio::spawn(
async move { chain_clone.check().await.unwrap() },
));
}
let mut results = Vec::new();
for handle in handles {
results.push(handle.await.unwrap());
}
for result in results {
assert_eq!(result, Decision::allowed_default());
}
let stats = chain.stats().await;
assert_eq!(stats.total_checks, 10);
}
#[test]
fn test_atomic_chain_stats_new() {
let stats = AtomicChainStats::new();
let snapshot = stats.snapshot();
assert_eq!(snapshot.total_checks, 0);
assert_eq!(snapshot.allowed_count, 0);
assert_eq!(snapshot.rejected_count, 0);
assert_eq!(snapshot.error_count, 0);
assert!(snapshot.node_rejections.is_empty());
}
#[test]
fn test_atomic_chain_stats_default() {
let stats = AtomicChainStats::default();
let snapshot = stats.snapshot();
assert_eq!(snapshot.total_checks, 0);
}
#[test]
fn test_atomic_chain_stats_increment_operations() {
let stats = AtomicChainStats::new();
stats.increment_total();
stats.increment_total();
stats.increment_total();
stats.increment_allowed();
stats.increment_allowed();
stats.increment_rejected();
stats.increment_error();
stats.increment_error();
stats.increment_error();
let snapshot = stats.snapshot();
assert_eq!(snapshot.total_checks, 3);
assert_eq!(snapshot.allowed_count, 2);
assert_eq!(snapshot.rejected_count, 1);
assert_eq!(snapshot.error_count, 3);
}
#[test]
fn test_atomic_chain_stats_node_rejections() {
let stats = AtomicChainStats::new();
stats.increment_node_rejection("node1");
stats.increment_node_rejection("node1");
stats.increment_node_rejection("node1");
stats.increment_node_rejection("node2");
let snapshot = stats.snapshot();
assert_eq!(snapshot.node_rejections.len(), 2);
let node1_count = snapshot
.node_rejections
.iter()
.find(|(id, _)| id == "node1")
.map(|(_, count)| *count)
.unwrap_or(0);
let node2_count = snapshot
.node_rejections
.iter()
.find(|(id, _)| id == "node2")
.map(|(_, count)| *count)
.unwrap_or(0);
assert_eq!(node1_count, 3);
assert_eq!(node2_count, 1);
}
#[test]
fn test_atomic_chain_stats_reset() {
let stats = AtomicChainStats::new();
stats.increment_total();
stats.increment_allowed();
stats.increment_node_rejection("node1");
stats.reset();
let snapshot = stats.snapshot();
assert_eq!(snapshot.total_checks, 0);
assert_eq!(snapshot.allowed_count, 0);
assert!(snapshot.node_rejections.is_empty());
}
#[test]
fn test_atomic_chain_stats_concurrent_increment() {
use std::sync::atomic::AtomicUsize;
use std::thread;
let stats = Arc::new(AtomicChainStats::new());
let counter = Arc::new(AtomicUsize::new(0));
let mut handles = vec![];
for _ in 0..100 {
let stats_clone = Arc::clone(&stats);
let counter_clone = Arc::clone(&counter);
handles.push(thread::spawn(move || {
stats_clone.increment_total();
stats_clone.increment_allowed();
counter_clone.fetch_add(1, Ordering::Relaxed);
}));
}
for handle in handles {
handle.join().unwrap();
}
let snapshot = stats.snapshot();
assert_eq!(snapshot.total_checks, 100);
assert_eq!(snapshot.allowed_count, 100);
}
#[test]
fn test_atomic_chain_stats_concurrent_node_rejections() {
use std::thread;
let stats = Arc::new(AtomicChainStats::new());
let mut handles = vec![];
for i in 0..50 {
let stats_clone = Arc::clone(&stats);
handles.push(thread::spawn(move || {
stats_clone.increment_node_rejection("node1");
stats_clone.increment_node_rejection(&format!("node{}", i % 5));
}));
}
for handle in handles {
handle.join().unwrap();
}
let snapshot = stats.snapshot();
let node1_count = snapshot
.node_rejections
.iter()
.find(|(id, _)| id == "node1")
.map(|(_, count)| *count)
.unwrap_or(0);
assert_eq!(node1_count, 60);
assert_eq!(snapshot.node_rejections.len(), 5);
}
#[tokio::test]
async fn test_decision_chain_stats_sync() {
let limiter = Arc::new(TokenBucketLimiter::new(5, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let chain = DecisionChain::with_dependencies(vec![node]);
for _ in 0..10 {
chain.check().await.unwrap();
}
let stats = chain.stats_sync();
assert_eq!(stats.total_checks, 10);
assert_eq!(stats.allowed_count, 5);
assert_eq!(stats.rejected_count, 5);
}
#[test]
fn test_decision_chain_reset_stats_sync() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node = DecisionNode::with_dependencies(
"node1".to_string(),
"Token Bucket".to_string(),
limiter,
100,
);
let chain = DecisionChain::with_dependencies(vec![node]);
chain.reset_stats_sync();
let stats = chain.stats_sync();
assert_eq!(stats.total_checks, 0);
}
#[test]
fn test_atomic_chain_stats_high_concurrency() {
use std::thread;
let stats = Arc::new(AtomicChainStats::new());
let mut handles = vec![];
for _ in 0..1000 {
let stats_clone = Arc::clone(&stats);
handles.push(thread::spawn(move || {
for _ in 0..100 {
stats_clone.increment_total();
stats_clone.increment_allowed();
}
}));
}
for handle in handles {
handle.join().unwrap();
}
let snapshot = stats.snapshot();
assert_eq!(snapshot.total_checks, 100_000);
assert_eq!(snapshot.allowed_count, 100_000);
}
#[tokio::test]
async fn test_decision_chain_high_concurrency_safety() {
let limiter = Arc::new(TokenBucketLimiter::new(10000, 1000));
let node = DecisionNode::with_dependencies(
"concurrent_node".to_string(),
"Concurrent Test".to_string(),
limiter,
100,
);
let chain = Arc::new(DecisionChain::with_dependencies(vec![node]));
let mut handles = vec![];
let barrier = Arc::new(tokio::sync::Barrier::new(100));
let start_signal = Arc::new(std::sync::atomic::AtomicBool::new(false));
for _ in 0..100 {
let chain_clone = Arc::clone(&chain);
let barrier_clone = Arc::clone(&barrier);
let start_signal_clone = Arc::clone(&start_signal);
handles.push(tokio::spawn(async move {
barrier_clone.wait().await;
while !start_signal_clone.load(std::sync::atomic::Ordering::SeqCst) {
std::hint::spin_loop();
}
let mut local_allowed = 0;
for _ in 0..100 {
if let Ok(Decision::Allowed(_)) = chain_clone.check().await {
local_allowed += 1;
}
}
local_allowed
}));
}
start_signal.store(true, std::sync::atomic::Ordering::SeqCst);
let mut total_allowed = 0;
for handle in handles {
total_allowed += handle.await.unwrap();
}
assert!(
total_allowed <= 10500,
"Total allowed: {}, expected <= 10500",
total_allowed
);
}
#[tokio::test]
async fn test_multi_node_chain_concurrent_safety() {
let limiter1 = Arc::new(TokenBucketLimiter::new(5000, 100));
let limiter2 = Arc::new(ShardedSlidingWindowLimiter::new(
Duration::from_secs(1),
5000,
));
let limiter3 = Arc::new(FixedWindowLimiter::new(Duration::from_secs(10), 5000));
let node1 = DecisionNode::with_dependencies(
"token_bucket".to_string(),
"Token Bucket".to_string(),
limiter1,
100,
);
let node2 = DecisionNode::with_dependencies(
"sliding_window".to_string(),
"Sliding Window".to_string(),
limiter2,
50,
);
let node3 = DecisionNode::with_dependencies(
"fixed_window".to_string(),
"Fixed Window".to_string(),
limiter3,
25,
);
let chain = Arc::new(DecisionChain::with_dependencies(vec![node1, node2, node3]));
let mut handles = vec![];
let barrier = Arc::new(tokio::sync::Barrier::new(50));
let start_signal = Arc::new(std::sync::atomic::AtomicBool::new(false));
for _ in 0..50 {
let chain_clone = Arc::clone(&chain);
let barrier_clone = Arc::clone(&barrier);
let start_signal_clone = Arc::clone(&start_signal);
handles.push(tokio::spawn(async move {
barrier_clone.wait().await;
while !start_signal_clone.load(std::sync::atomic::Ordering::SeqCst) {
std::hint::spin_loop();
}
let mut local_allowed = 0;
for _ in 0..100 {
if let Ok(Decision::Allowed(_)) = chain_clone.check().await {
local_allowed += 1;
}
}
local_allowed
}));
}
start_signal.store(true, std::sync::atomic::Ordering::SeqCst);
let mut total_allowed = 0;
for handle in handles {
total_allowed += handle.await.unwrap();
}
assert!(
total_allowed <= 5500,
"Total allowed: {}, expected <= 5500",
total_allowed
);
}
#[tokio::test]
async fn test_chain_stats_concurrent_safety() {
let limiter = Arc::new(TokenBucketLimiter::new(10000, 1000));
let node = DecisionNode::with_dependencies(
"stats_node".to_string(),
"Stats Test".to_string(),
limiter,
100,
);
let chain = Arc::new(DecisionChain::with_dependencies(vec![node]));
let mut handles = vec![];
for _ in 0..100 {
let chain_clone = Arc::clone(&chain);
handles.push(tokio::spawn(async move {
for _ in 0..10 {
let _ = chain_clone.check().await;
let _ = chain_clone.stats_sync();
}
}));
}
for handle in handles {
handle.await.unwrap();
}
let stats = chain.stats_sync();
assert!(stats.total_checks <= 10000);
}
#[tokio::test]
async fn test_chain_no_deadlock() {
let limiter = Arc::new(TokenBucketLimiter::new(100000, 10000));
let node = DecisionNode::with_dependencies(
"deadlock_test".to_string(),
"Deadlock Test".to_string(),
limiter,
100,
);
let chain = Arc::new(DecisionChain::with_dependencies(vec![node]));
let mut handles = vec![];
for _ in 0..500 {
let chain_clone = Arc::clone(&chain);
handles.push(tokio::spawn(async move {
for _ in 0..50 {
let _ = chain_clone.check().await;
}
}));
}
let result = tokio::time::timeout(Duration::from_secs(10), async {
for handle in handles {
let _ = handle.await;
}
})
.await;
assert!(result.is_ok(), "Test timed out - possible deadlock");
}
#[tokio::test]
async fn test_node_enable_disable_concurrent_safety() {
let limiter = Arc::new(TokenBucketLimiter::new(10000, 1000));
let node = DecisionNode::with_dependencies(
"toggle_node".to_string(),
"Toggle Test".to_string(),
limiter,
100,
);
let chain = Arc::new(tokio::sync::RwLock::new(DecisionChain::with_dependencies(
vec![node],
)));
let mut handles = vec![];
for i in 0..100 {
let chain_clone = Arc::clone(&chain);
handles.push(tokio::spawn(async move {
if i % 2 == 0 {
let chain = chain_clone.read().await;
let _ = chain.check().await;
} else {
let mut chain = chain_clone.write().await;
if i % 4 == 1 {
chain.disable_node("toggle_node");
} else {
chain.enable_node("toggle_node");
}
}
}));
}
for handle in handles {
handle.await.unwrap();
}
let chain = chain.read().await;
let stats = chain.stats_sync();
assert!(stats.total_checks <= 10000);
}
#[tokio::test]
async fn test_empty_chain_boundary() {
let chain = DecisionChain::with_dependencies(vec![]);
for _ in 0..100 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let stats = chain.stats_sync();
assert_eq!(stats.total_checks, 100);
assert_eq!(stats.allowed_count, 100);
}
#[tokio::test]
async fn test_all_nodes_disabled_boundary() {
let limiter = Arc::new(TokenBucketLimiter::new(0, 1)); let node = DecisionNode::with_dependencies(
"disabled_node".to_string(),
"Disabled Node".to_string(),
limiter,
100,
)
.with_enabled(false);
let chain = DecisionChain::with_dependencies(vec![node]);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
#[tokio::test]
async fn test_chain_cost_boundary() {
let limiter = Arc::new(TokenBucketLimiter::new(100, 1));
let node = DecisionNode::with_dependencies(
"cost_node".to_string(),
"Cost Test".to_string(),
limiter,
100,
)
.with_cost(50);
let chain = DecisionChain::with_dependencies(vec![node]);
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
}
#[test]
fn test_decision_node_new_backward_compat() {
let limiter = Arc::new(TokenBucketLimiter::new(100, 10));
let node = DecisionNode::new("node1".to_string(), "new method".to_string(), limiter, 100);
assert_eq!(node.id, "node1");
assert_eq!(node.name, "new method");
assert_eq!(node.priority, 100);
assert!(node.enabled);
assert!(node.short_circuit);
assert_eq!(node.cost, 1);
}
#[test]
fn test_decision_chain_builder_method() {
let builder = DecisionChain::builder();
let chain = builder.build();
assert_eq!(chain.node_count(), 0);
}
#[test]
fn test_decision_chain_builder_default_trait() {
let _builder = DecisionChainBuilder::default();
}
struct ErrorLimiter;
#[async_trait]
impl Limiter for ErrorLimiter {
async fn allow(&self, _cost: u64) -> Result<bool, LimiteronError> {
Err(LimiteronError::LimitError("limit error".to_string()))
}
}
#[tokio::test]
async fn test_decision_chain_check_error_path() {
let node = DecisionNode::with_dependencies(
"error_node".to_string(),
"Error Limiter".to_string(),
Arc::new(ErrorLimiter),
100,
);
let chain = DecisionChain::with_dependencies(vec![node]);
let result = chain.check().await;
assert!(result.is_err());
let stats = chain.stats_sync();
assert_eq!(stats.total_checks, 1);
assert_eq!(stats.error_count, 1);
}
#[tokio::test]
async fn test_decision_chain_error_path_second_node() {
let ok = Arc::new(TokenBucketLimiter::new(100, 10));
let ok_node = DecisionNode::with_dependencies("ok".to_string(), "OK".to_string(), ok, 100);
let err_node = DecisionNode::with_dependencies(
"err".to_string(),
"Error".to_string(),
Arc::new(ErrorLimiter),
50,
);
let chain = DecisionChain::with_dependencies(vec![ok_node, err_node]);
let result = chain.check().await;
assert!(result.is_err());
let stats = chain.stats_sync();
assert_eq!(stats.total_checks, 1);
assert_eq!(stats.error_count, 1);
}
#[tokio::test]
async fn test_decision_chain_non_short_circuit_final() {
let pass = Arc::new(MockLimiter::new(true));
let limited = Arc::new(TokenBucketLimiter::new(3, 1));
let node1 = DecisionNode::with_dependencies(
"pass".to_string(),
"Pass Through".to_string(),
pass,
100,
);
let node2 = DecisionNode::with_dependencies(
"limited".to_string(),
"Limited".to_string(),
limited,
50,
)
.with_short_circuit(false);
let chain = DecisionChain::with_dependencies(vec![node1, node2]);
for _ in 0..3 {
let decision = chain.check().await.unwrap();
assert_eq!(decision, Decision::allowed_default());
}
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
if let Decision::Rejected(meta) = decision {
assert!(meta.reason.contains("Limited"));
}
let stats = chain.stats_sync();
assert_eq!(stats.total_checks, 5);
assert_eq!(stats.rejected_count, 1);
}
#[tokio::test]
async fn test_decision_chain_enable_node_not_found() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node =
DecisionNode::with_dependencies("node1".to_string(), "Node".to_string(), limiter, 100);
let mut chain = DecisionChain::with_dependencies(vec![node]);
assert!(!chain.enable_node("nonexistent"));
}
#[tokio::test]
async fn test_decision_chain_disable_node_not_found() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node =
DecisionNode::with_dependencies("node1".to_string(), "Node".to_string(), limiter, 100);
let mut chain = DecisionChain::with_dependencies(vec![node]);
assert!(!chain.disable_node("nonexistent"));
}
#[tokio::test]
async fn test_decision_chain_set_short_circuit_found() {
let spy = Arc::new(SpyLimiter::new());
let reject = Arc::new(MockLimiter::new(true));
let node1 = DecisionNode::with_dependencies(
"node1".to_string(),
"Mock".to_string(),
reject.clone(),
100,
);
let node2 = DecisionNode::with_dependencies(
"node2".to_string(),
"Spy".to_string(),
spy.clone(),
50,
);
let mut chain = DecisionChain::with_dependencies(vec![node1, node2]);
chain.check().await.unwrap();
assert_eq!(spy.calls.load(std::sync::atomic::Ordering::SeqCst), 1);
assert!(chain.set_short_circuit("node1", true));
reject.set_allowed(false);
let decision = chain.check().await.unwrap();
assert!(matches!(decision, Decision::Rejected(_)));
assert_eq!(spy.calls.load(std::sync::atomic::Ordering::SeqCst), 1);
}
#[tokio::test]
async fn test_decision_chain_set_short_circuit_not_found() {
let limiter = Arc::new(TokenBucketLimiter::new(10, 1));
let node =
DecisionNode::with_dependencies("node1".to_string(), "Node".to_string(), limiter, 100);
let mut chain = DecisionChain::with_dependencies(vec![node]);
assert!(!chain.set_short_circuit("nonexistent", true));
}
}