andwati/telecom-platform
1
0
Fork 0
mirror of https://github.com/andwati/telecom-platform.git synced 2026-10-18 11:02:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Comment out circuit breaker implementation and remove thiserror dependency

Browse Source 
- Remove thiserror 2.0.18 dependency from Cargo.toml
- Comment out CircuitBreakerError import and all circuit breaker execute calls in credit_management.rs
- Comment out redis_circuit_breaker and postgres_circuit_breaker fields in ChargingEngine struct
- Comment out circuit breaker initialization code in engine.rs
- Comment out circuit_breaker field and initialization in RatingPlansRepo
- Remove circuit breaker
This commit is contained in:
nutcas3
2026-04-26 01:15:58 +03:00
parent b6bcad067b
commit d5ebb3bfd8
8 changed files with 103 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/charging-engine/Cargo.toml
-1
View File
@@ -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"
apps/charging-engine/src/charging/credit_management.rs
+46 -46
View File
@@ -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<u64> {
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<u64> {
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<u64> {
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<bool> {
@@ -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<bool> {
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<Option<SubscriberAccount>> {
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,
// })
}
}
apps/charging-engine/src/charging/engine.rs
+7 -5
View File
@@ -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)),
})
}
apps/charging-engine/src/charging/rating_plans_repo.rs
+24 -33
View File
@@ -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<Option<RatingPlan>> {
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<HashMap<String, RatingPlan>> {
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<bool> {
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,
// })
}
}
apps/charging-engine/src/circuit_breaker.rs
+18 -11
View File
@@ -9,17 +9,23 @@ pub enum CircuitState {
HalfOpen,
}
#[derive(Debug, thiserror::Error)]
pub enum CircuitBreakerError<E>
where
E: std::error::Error + Send + Sync + 'static,
{
#[error("Circuit breaker is open")]
#[derive(Debug)]
pub enum CircuitBreakerError<E> {
Open,
#[error("Wrapped error: {0}")]
Inner(#[from] E),
Inner(E),
}
impl<E: std::fmt::Display> std::fmt::Display for CircuitBreakerError<E> {
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<E: std::error::Error + Send + Sync + 'static> std::error::Error for CircuitBreakerError<E> {}
#[derive(Clone)]
pub struct CircuitBreaker {
state: Arc<RwLock<CircuitState>>,
@@ -40,9 +46,10 @@ impl CircuitBreaker {
}
}
pub async fn execute<F, T, E>(&self, operation: F) -> Result<T, CircuitBreakerError<E>>
pub async fn execute<F, Fut, T, E>(&self, operation: F) -> Result<T, CircuitBreakerError<E>>
where
F: std::future::Future<Output = Result<T, E>>,
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<T, E>>,
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) => {
apps/charging-engine/src/lib.rs
+2 -2
View File
@@ -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;
apps/charging-engine/src/rate_limit.rs
+3 -6
View File
@@ -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<SmartIpKeyExtractor, StateInformationMiddleware<governor::clock::QuantaClock>> {
pub fn create_rate_limiter() -> tower_governor::GovernorLayer<SmartIpKeyExtractor, governor::state::InMemoryState, governor::clock::DefaultClock> {
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<SmartIpKeyExtracto
.finish()
.unwrap();
tower_governor::GovernorLayer {
config: Arc::new(config),
}
tower_governor::GovernorLayer::new(config)
}
pub struct RateLimitResponse;
apps/charging-engine/src/routes.rs
+3 -3
View File
@@ -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)
}