From a4aec9431417baacd0c14267976da93f569ba5df Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Sun, 26 Apr 2026 02:13:07 +0300 Subject: [PATCH] refactor: Re-enable circuit breaker implementation in charging-engine credit management and rating plans - Uncomment CircuitBreaker and CircuitBreakerError imports in credit_management.rs and rating_plans_repo.rs - Re-enable redis_circuit_breaker execute calls for all Redis operations (get_balance, add_credit, deduct_credit, block_user, unblock_user, is_user_blocked, get_subscriber_account, update_subscriber_account, record_usage_event) - Re-enable postgres circuit_breaker execute calls for all database operations --- .../src/charging/credit_management.rs | 94 +++++++++---------- .../src/charging/rating_plans_repo.rs | 48 +++++----- apps/charging-engine/src/lib.rs | 5 +- apps/charging-engine/src/routes.rs | 6 +- 4 files changed, 76 insertions(+), 77 deletions(-) diff --git a/apps/charging-engine/src/charging/credit_management.rs b/apps/charging-engine/src/charging/credit_management.rs index c356b9b..217539b 100644 --- a/apps/charging-engine/src/charging/credit_management.rs +++ b/apps/charging-engine/src/charging/credit_management.rs @@ -1,12 +1,12 @@ use tracing::{info, warn, debug}; use super::types::{SubscriberAccount, UsageEvent}; -// use crate::circuit_breaker::CircuitBreakerError; +use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError}; use crate::errors::{ChargingError, ChargingResult}; impl crate::charging::ChargingEngine { pub async fn get_balance(&self, ip: &str) -> ChargingResult { - // // // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -14,16 +14,16 @@ impl crate::charging::ChargingEngine { let balance: u64 = redis::AsyncCommands::get(&mut conn, &key).await .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - debug!("Retrieved balance {} for IP: {}", balance, ip); + info!("Retrieved balance for IP: {}, balance: {} bytes", ip, balance); Ok(balance) - // }).await.map_err(|e| match e {$ - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn add_credit(&self, ip: &str, bytes_to_add: u64) -> ChargingResult { - // // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -51,14 +51,14 @@ impl crate::charging::ChargingEngine { info!("Added {} bytes credit to IP: {}, new balance: {}", bytes_to_add, ip, new_balance); Ok(new_balance) - // }).await.map_err(|e| match e {$ - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn deduct_credit(&self, ip: &str, bytes_used: u64) -> ChargingResult { - // // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -80,10 +80,10 @@ impl crate::charging::ChargingEngine { info!("Deducted {} bytes credit from IP: {}, new balance: {}", bytes_used, ip, new_balance); Ok(new_balance) - // }).await.map_err(|e| match e {$ - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn check_credit(&self, ip: &str, bytes_requested: u64) -> ChargingResult { @@ -92,7 +92,7 @@ impl crate::charging::ChargingEngine { } pub async fn block_user(&self, ip: &str) -> ChargingResult<()> { - // // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -116,14 +116,14 @@ impl crate::charging::ChargingEngine { warn!("User IP: {} has been blocked for {} seconds", ip, block_expiration); Ok(()) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn unblock_user(&self, ip: &str) -> ChargingResult<()> { - // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -133,14 +133,14 @@ impl crate::charging::ChargingEngine { info!("User IP: {} has been unblocked", ip); Ok(()) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn is_user_blocked(&self, ip: &str) -> ChargingResult { - // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -149,14 +149,14 @@ impl crate::charging::ChargingEngine { .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; Ok(blocked.is_some()) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn get_subscriber_account(&self, imsi: &str) -> ChargingResult> { - // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -166,14 +166,14 @@ impl crate::charging::ChargingEngine { info!("Retrieved subscriber account for IMSI: {}", imsi); Ok(account) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn update_subscriber_account(&self, account: &SubscriberAccount) -> ChargingResult<()> { - // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -183,14 +183,14 @@ impl crate::charging::ChargingEngine { debug!("Updated subscriber account for IMSI: {}", account.imsi); Ok(()) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } pub async fn record_usage_event(&self, event: &UsageEvent) -> ChargingResult<()> { - // self.redis_circuit_breaker.execute(|| async { + self.redis_circuit_breaker.execute(|| async { let mut conn = self.redis_client.get_multiplexed_async_connection().await .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; @@ -200,9 +200,9 @@ impl crate::charging::ChargingEngine { info!("Recorded usage event for IMSI: {}, session: {}", event.imsi, event.session_id); Ok(()) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } } diff --git a/apps/charging-engine/src/charging/rating_plans_repo.rs b/apps/charging-engine/src/charging/rating_plans_repo.rs index a29d945..b2d447b 100644 --- a/apps/charging-engine/src/charging/rating_plans_repo.rs +++ b/apps/charging-engine/src/charging/rating_plans_repo.rs @@ -9,14 +9,14 @@ use sqlx::postgres::{PgPool, PgPoolOptions}; use sqlx::Row; use super::types::RatingPlan; -// use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError}; +use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError}; use crate::errors::{ChargingError, ChargingResult}; /// Repository for rating plans backed by Postgres. #[derive(Clone)] pub struct RatingPlansRepo { pool: PgPool, - // circuit_breaker: CircuitBreaker, + circuit_breaker: CircuitBreaker, } impl RatingPlansRepo { @@ -28,7 +28,7 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::InternalError(e.to_string()))?; - // let circuit_breaker = CircuitBreaker::new(5, std::time::Duration::from_secs(60)); + let circuit_breaker = CircuitBreaker::new(5, std::time::Duration::from_secs(60)); // Idempotent schema for rating plans. api-server's GORM migration also // creates a compatible `rating_plans` table; the column set here is @@ -55,7 +55,7 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; - Ok(Self { pool }) + Ok(Self { pool, circuit_breaker }) } /// Seed the canonical basic/premium/enterprise plans if the table is empty. @@ -106,7 +106,7 @@ impl RatingPlansRepo { /// Fetch a single plan by id. pub async fn get(&self, plan_id: &str) -> ChargingResult> { let plan_id = plan_id.to_string(); - // self.circuit_breaker.execute(|| async move { + self.circuit_breaker.execute(|| async move { let row = sqlx::query( r#" SELECT plan_id, name, data_rate, voice_rate, sms_rate, @@ -121,15 +121,15 @@ impl RatingPlansRepo { .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; Ok(row.map(row_to_plan)) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } /// Fetch all active plans as a HashMap for fast lookup. pub async fn list_map(&self) -> ChargingResult> { - // self.circuit_breaker.execute(|| async { + self.circuit_breaker.execute(|| async { let rows = sqlx::query( r#" SELECT plan_id, name, data_rate, voice_rate, sms_rate, @@ -148,10 +148,10 @@ impl RatingPlansRepo { out.insert(plan.plan_id.clone(), plan); } Ok(out) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } /// List all active plans. @@ -178,7 +178,7 @@ impl RatingPlansRepo { /// Insert or update a plan. pub async fn upsert(&self, plan: &RatingPlan) -> ChargingResult<()> { let plan = plan.clone(); - // self.circuit_breaker.execute(|| async move { + self.circuit_breaker.execute(|| async move { sqlx::query( r#" INSERT INTO rating_plans @@ -211,16 +211,16 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; Ok(()) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } /// Soft-delete a plan by marking it inactive. pub async fn deactivate(&self, plan_id: &str) -> ChargingResult { let plan_id = plan_id.to_string(); - // self.circuit_breaker.execute(|| async move { + self.circuit_breaker.execute(|| async move { let result = sqlx::query( r#"UPDATE rating_plans SET is_active = FALSE, updated_at = NOW() WHERE plan_id = $1"#, ) @@ -229,10 +229,10 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; Ok(result.rows_affected() > 0) - // }).await.map_err(|e| match e { - // CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), - // CircuitBreakerError::Inner(e) => e, - // }) + }).await.map_err(|e| match e { + CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()), + CircuitBreakerError::Inner(e) => e, + }) } } diff --git a/apps/charging-engine/src/lib.rs b/apps/charging-engine/src/lib.rs index 450c52c..be77c33 100644 --- a/apps/charging-engine/src/lib.rs +++ b/apps/charging-engine/src/lib.rs @@ -1,9 +1,8 @@ pub mod auth; pub mod charging; -// pub mod circuit_breaker; -pub mod errors; +pub mod circuit_breaker; +pub mod rate_limit; pub mod handlers; -// pub mod rate_limit; pub mod routes; pub mod models; pub mod monitoring; diff --git a/apps/charging-engine/src/routes.rs b/apps/charging-engine/src/routes.rs index fc9fc6c..ab944e3 100644 --- a/apps/charging-engine/src/routes.rs +++ b/apps/charging-engine/src/routes.rs @@ -5,7 +5,7 @@ use axum::{ use tower_http::cors::{Any, CorsLayer}; use crate::auth::auth_middleware; -// use crate::rate_limit::create_rate_limiter; +use crate::rate_limit::create_rate_limiter; use crate::handlers::{ add_credit, add_rating_plan, block_user, check_credit, deduct_credit, detailed_health_check, engine_start, engine_stop, engine_uptime, get_balance, get_error_stats, get_performance_metrics, @@ -21,7 +21,7 @@ pub fn create_router(state: AppState) -> Router { .allow_methods(Any) .allow_headers(Any); - // let rate_limiter = create_rate_limiter(); + let rate_limiter = create_rate_limiter(); Router::new() // Public routes (no auth required - for packet-gateway integration) @@ -54,7 +54,7 @@ pub fn create_router(state: AppState) -> Router { .route("/v1/usage/process", post(process_usage)) .route("/v1/invoice/:imsi/:period", get(generate_invoice)) .layer(cors) - // .layer(rate_limiter) + .layer(rate_limiter) .route_layer(axum::middleware::from_fn_with_state(state.auth_config.clone(), auth_middleware)) .with_state(state) }