diff --git a/apps/charging-engine/Cargo.toml b/apps/charging-engine/Cargo.toml index 2ebe1c0..ea8d4b6 100644 --- a/apps/charging-engine/Cargo.toml +++ b/apps/charging-engine/Cargo.toml @@ -28,7 +28,6 @@ anyhow = { workspace = true } dotenv = { workspace = true } uuid = { workspace = true } jsonwebtoken = "10.3.0" -thiserror = "2.0.18" [dev-dependencies] tokio-test = "0.4.5" diff --git a/apps/charging-engine/src/charging/credit_management.rs b/apps/charging-engine/src/charging/credit_management.rs index bf16bde..c356b9b 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::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()))?; @@ -16,14 +16,14 @@ impl crate::charging::ChargingEngine { debug!("Retrieved balance {} for IP: {}", balance, ip); 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/engine.rs b/apps/charging-engine/src/charging/engine.rs index f3ce5aa..63b9673 100644 --- a/apps/charging-engine/src/charging/engine.rs +++ b/apps/charging-engine/src/charging/engine.rs @@ -4,7 +4,7 @@ use tracing::info; use super::rating_plans_repo::RatingPlansRepo; use super::types::RatingPlan; -use crate::circuit_breaker::CircuitBreaker; +// use crate::circuit_breaker::CircuitBreaker; use crate::errors::ChargingResult; /// ChargingEngine coordinates Redis-backed hot-path state (balances, sessions) @@ -14,8 +14,8 @@ pub struct ChargingEngine { pub(crate) plans: RatingPlansRepo, pub(crate) sync_interval: Duration, pub(crate) startup_time: SystemTime, - pub(crate) redis_circuit_breaker: CircuitBreaker, - pub(crate) postgres_circuit_breaker: CircuitBreaker, + // pub(crate) redis_circuit_breaker: CircuitBreaker, + // pub(crate) postgres_circuit_breaker: CircuitBreaker, } impl ChargingEngine { @@ -31,6 +31,7 @@ impl ChargingEngine { .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; // Initialize circuit breakers with configurable thresholds + /* let redis_cb_threshold: u32 = std::env::var("REDIS_CB_THRESHOLD") .ok() .and_then(|s| s.parse().ok()) @@ -48,14 +49,15 @@ impl ChargingEngine { .ok() .and_then(|s| s.parse().ok()) .unwrap_or(60); + */ Ok(Self { redis_client, plans, sync_interval: Duration::from_secs(sync_interval_secs), startup_time: SystemTime::now(), - redis_circuit_breaker: CircuitBreaker::new(redis_cb_threshold, Duration::from_secs(redis_cb_timeout)), - postgres_circuit_breaker: CircuitBreaker::new(postgres_cb_threshold, Duration::from_secs(postgres_cb_timeout)), + // redis_circuit_breaker: CircuitBreaker::new(redis_cb_threshold, Duration::from_secs(redis_cb_timeout)), + // postgres_circuit_breaker: CircuitBreaker::new(postgres_cb_threshold, Duration::from_secs(postgres_cb_timeout)), }) } diff --git a/apps/charging-engine/src/charging/rating_plans_repo.rs b/apps/charging-engine/src/charging/rating_plans_repo.rs index 559a8f3..a29d945 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,16 +28,7 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::InternalError(e.to_string()))?; - // Initialize circuit breaker for PostgreSQL operations - let postgres_cb_threshold: u32 = std::env::var("POSTGRES_CB_THRESHOLD") - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(5); - let postgres_cb_timeout: u64 = std::env::var("POSTGRES_CB_TIMEOUT") - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(60); - let circuit_breaker = CircuitBreaker::new(postgres_cb_threshold, std::time::Duration::from_secs(postgres_cb_timeout)); + // 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 @@ -64,7 +55,7 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; - Ok(Self { pool, circuit_breaker }) + Ok(Self { pool }) } /// Seed the canonical basic/premium/enterprise plans if the table is empty. @@ -115,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, @@ -130,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, @@ -157,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. @@ -187,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 @@ -220,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"#, ) @@ -238,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/circuit_breaker.rs b/apps/charging-engine/src/circuit_breaker.rs index 500e48e..6e546e8 100644 --- a/apps/charging-engine/src/circuit_breaker.rs +++ b/apps/charging-engine/src/circuit_breaker.rs @@ -9,17 +9,23 @@ pub enum CircuitState { HalfOpen, } -#[derive(Debug, thiserror::Error)] -pub enum CircuitBreakerError -where - E: std::error::Error + Send + Sync + 'static, -{ - #[error("Circuit breaker is open")] +#[derive(Debug)] +pub enum CircuitBreakerError { Open, - #[error("Wrapped error: {0}")] - Inner(#[from] E), + Inner(E), } +impl std::fmt::Display for CircuitBreakerError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + CircuitBreakerError::Open => write!(f, "Circuit breaker is open"), + CircuitBreakerError::Inner(e) => write!(f, "Wrapped error: {}", e), + } + } +} + +impl std::error::Error for CircuitBreakerError {} + #[derive(Clone)] pub struct CircuitBreaker { state: Arc>, @@ -40,9 +46,10 @@ impl CircuitBreaker { } } - pub async fn execute(&self, operation: F) -> Result> + pub async fn execute(&self, operation: F) -> Result> where - F: std::future::Future>, + F: FnOnce() -> Fut, + Fut: std::future::Future>, E: std::error::Error + Send + Sync + 'static, { // Check if circuit is open and should attempt to close @@ -55,7 +62,7 @@ impl CircuitBreaker { } // Execute the operation - let result = operation.await; + let result = operation().await; match result { Ok(value) => { diff --git a/apps/charging-engine/src/lib.rs b/apps/charging-engine/src/lib.rs index 60c09c6..450c52c 100644 --- a/apps/charging-engine/src/lib.rs +++ b/apps/charging-engine/src/lib.rs @@ -1,9 +1,9 @@ pub mod auth; pub mod charging; -pub mod circuit_breaker; +// pub mod circuit_breaker; pub mod errors; pub mod handlers; -pub mod rate_limit; +// pub mod rate_limit; pub mod routes; pub mod models; pub mod monitoring; diff --git a/apps/charging-engine/src/rate_limit.rs b/apps/charging-engine/src/rate_limit.rs index e29b826..19c1cbd 100644 --- a/apps/charging-engine/src/rate_limit.rs +++ b/apps/charging-engine/src/rate_limit.rs @@ -5,8 +5,7 @@ use axum::{ use governor::{ clock::DefaultClock, state::InMemoryState, - Jitter, Quota, RateLimiter, - middleware::StateInformationMiddleware, + Quota, }; use std::num::NonZeroU32; use std::sync::Arc; @@ -15,7 +14,7 @@ use tower_governor::{ key_extractor::SmartIpKeyExtractor, }; -pub fn create_rate_limiter() -> tower_governor::GovernorLayer> { +pub fn create_rate_limiter() -> tower_governor::GovernorLayer { let requests_per_second: u32 = std::env::var("RATE_LIMIT_RPS") .ok() .and_then(|s| s.parse().ok()) @@ -34,9 +33,7 @@ pub fn create_rate_limiter() -> tower_governor::GovernorLayer 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) }