//! Backtesting Service Proxy - Zero-copy gRPC forwarding //! //! This module implements a high-performance proxy for the backtesting service. //! Design goals: //! - <10μs routing overhead //! - Zero-copy forwarding where possible //! - Connection pooling via tonic::transport::Channel //! - Circuit breaker for backend failures //! - Health checking of backend service use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::RwLock; use tonic::transport::{Certificate, ClientTlsConfig, Identity}; use tonic::{Request, Response, Status}; use tonic_health::pb::health_client::HealthClient; use tracing::{debug, error, info, warn, instrument}; // Import generated protobuf types from build.rs use crate::foxhunt::tli::{ backtesting_service_client::BacktestingServiceClient, backtesting_service_server::BacktestingService, BacktestProgressEvent, GetBacktestResultsRequest, GetBacktestResultsResponse, GetBacktestStatusRequest, GetBacktestStatusResponse, ListBacktestsRequest, ListBacktestsResponse, // Request/Response types StartBacktestRequest, StartBacktestResponse, StopBacktestRequest, StopBacktestResponse, SubscribeBacktestProgressRequest, }; /// Health check state for circuit breaker #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum HealthState { Healthy, Degraded, Unhealthy, } /// Health checker for backend service with circuit breaker pub struct HealthChecker { state: RwLock, consecutive_failures: RwLock, last_health_check: RwLock, failure_threshold: u32, health_check_interval: Duration, } impl HealthChecker { pub fn new(failure_threshold: u32, health_check_interval: Duration) -> Self { Self { state: RwLock::new(HealthState::Healthy), consecutive_failures: RwLock::new(0), last_health_check: RwLock::new(Instant::now()), failure_threshold, health_check_interval, } } /// Record a successful request pub async fn record_success(&self) { let mut failures = self.consecutive_failures.write().await; *failures = 0; let mut state = self.state.write().await; if *state != HealthState::Healthy { info!("Backtesting service backend recovered to healthy state"); *state = HealthState::Healthy; } } /// Record a failed request pub async fn record_failure(&self) { let mut failures = self.consecutive_failures.write().await; *failures += 1; let mut state = self.state.write().await; if *failures >= self.failure_threshold && *state != HealthState::Unhealthy { error!( "Backtesting service backend marked as unhealthy after {} consecutive failures", failures ); *state = HealthState::Unhealthy; } else if *failures >= self.failure_threshold / 2 && *state == HealthState::Healthy { warn!( "Backtesting service backend degraded after {} failures", failures ); *state = HealthState::Degraded; } } /// Check if backend is healthy enough to accept requests pub async fn is_healthy(&self) -> bool { let state = self.state.read().await; *state != HealthState::Unhealthy } /// Get current health state pub async fn get_state(&self) -> HealthState { *self.state.read().await } /// Perform proactive gRPC health check using tonic-health pub async fn perform_health_check(&self, channel: &tonic::transport::Channel) { // Check if enough time has passed since last check let now = Instant::now(); let should_check = { let last_check = self.last_health_check.read().await; now.duration_since(*last_check) >= self.health_check_interval }; if !should_check { return; } // Update last check time { let mut last_check = self.last_health_check.write().await; *last_check = now; } // Perform gRPC health check let mut health_client = HealthClient::new(channel.clone()); let request = tonic_health::pb::HealthCheckRequest { service: "foxhunt.tli.BacktestingService".to_string(), }; match health_client.check(request).await { Ok(response) => { let status = response.into_inner().status; if status == tonic_health::pb::health_check_response::ServingStatus::Serving as i32 { self.record_success().await; debug!("Backtesting service health check: SERVING"); } else { self.record_failure().await; warn!( "Backtesting service health check: NOT_SERVING (status: {})", status ); } }, Err(e) => { self.record_failure().await; error!("Backtesting service health check failed: {}", e); }, } } } /// Backtesting service proxy with zero-copy forwarding pub struct BacktestingServiceProxy { /// Client connection to backend service /// tonic::transport::Channel is cheap to clone and reuses connections client: BacktestingServiceClient, /// Health checker with circuit breaker health_checker: Arc, /// Backend service URL for logging backend_url: String, /// Underlying channel for health checks channel: tonic::transport::Channel, } impl BacktestingServiceProxy { /// Create a new backtesting service proxy with TLS/mTLS support /// /// # Arguments /// * `backend_url` - URL of the backend backtesting service (e.g., "https://localhost:50053") /// * `ca_cert_path` - Path to CA certificate for server verification /// * `client_cert_path` - Path to client certificate for mTLS authentication /// * `client_key_path` - Path to client private key for mTLS authentication /// /// # Returns /// * `Result` - Proxy instance or connection error pub async fn new( backend_url: &str, ca_cert_path: Option<&str>, client_cert_path: Option<&str>, client_key_path: Option<&str>, ) -> Result> { info!( "Connecting to backtesting service backend at {}", backend_url ); // Establish connection to backend service // tonic::transport::Channel automatically handles: // - Connection pooling // - Keep-alive // - Load balancing (if multiple endpoints provided) let mut endpoint = tonic::transport::Endpoint::from_shared(backend_url.to_string())? .connect_timeout(Duration::from_secs(5)) .timeout(Duration::from_secs(30)) .tcp_keepalive(Some(Duration::from_secs(60))) .http2_keep_alive_interval(Duration::from_secs(30)) .keep_alive_while_idle(true); // Configure TLS if certificate paths provided (HTTPS URLs) if backend_url.starts_with("https://") { match (ca_cert_path, client_cert_path, client_key_path) { (Some(ca_path), Some(cert_path), Some(key_path)) => { info!("Configuring TLS with mTLS (client certificates)"); info!(" CA cert: {}", ca_path); info!(" Client cert: {}", cert_path); info!(" Client key: {}", key_path); // Load certificates from files info!("Reading CA certificate..."); let ca_pem = tokio::fs::read_to_string(ca_path).await.map_err(|e| { error!("Failed to read CA certificate at {}: {}", ca_path, e); format!("Failed to read CA certificate at {}: {}", ca_path, e) })?; info!("CA certificate loaded ({} bytes)", ca_pem.len()); info!("Reading client certificate..."); let client_cert_pem = tokio::fs::read_to_string(cert_path).await.map_err(|e| { error!("Failed to read client certificate at {}: {}", cert_path, e); format!("Failed to read client certificate at {}: {}", cert_path, e) })?; info!( "Client certificate loaded ({} bytes)", client_cert_pem.len() ); info!("Reading client key..."); let client_key_pem = tokio::fs::read_to_string(key_path).await.map_err(|e| { error!("Failed to read client key at {}: {}", key_path, e); format!("Failed to read client key at {}: {}", key_path, e) })?; info!("Client key loaded ({} bytes)", client_key_pem.len()); // Create TLS configuration with mTLS (client authentication) // Extract hostname from backend_url for SNI (Server Name Indication) let hostname = if let Some(host) = backend_url.strip_prefix("https://") { host.split(':').next().unwrap_or("backtesting_service") } else { "backtesting_service" }; let tls_config = ClientTlsConfig::new() .ca_certificate(Certificate::from_pem(&ca_pem)) .identity(Identity::from_pem(&client_cert_pem, &client_key_pem)) .domain_name(hostname); // Use actual hostname for SNI info!("TLS SNI hostname: {}", hostname); endpoint = endpoint.tls_config(tls_config).map_err(|e| { error!("Failed to apply TLS configuration: {:?}", e); error!(" TLS config error: {}", e); e })?; info!("TLS configuration with mTLS applied successfully"); }, (Some(ca_path), None, None) => { info!("Configuring TLS with server verification only (no client cert)"); let ca_pem = tokio::fs::read_to_string(ca_path).await.map_err(|e| { format!("Failed to read CA certificate at {}: {}", ca_path, e) })?; let tls_config = ClientTlsConfig::new() .ca_certificate(Certificate::from_pem(&ca_pem)) .domain_name("foxhunt-services"); // Match server cert CN endpoint = endpoint.tls_config(tls_config)?; info!("TLS configuration with server verification applied successfully"); }, _ => { warn!( "HTTPS URL provided but certificate paths incomplete - connection may fail" ); warn!( " CA: {:?}, Client cert: {:?}, Client key: {:?}", ca_cert_path, client_cert_path, client_key_path ); }, } } else { info!("Using HTTP (no TLS) for backtesting service connection"); } let channel = endpoint.connect().await.map_err(|e| { error!("Failed to connect to backtesting service: {:?}", e); error!(" Error details: {}", e); e })?; info!("Successfully established channel connection to backtesting service"); let client = BacktestingServiceClient::new(channel.clone()); // Initialize health checker with circuit breaker let health_checker = Arc::new(HealthChecker::new( 5_u32, // failure_threshold: 5 consecutive failures Duration::from_secs(10), // health_check_interval: 10 seconds )); info!("Successfully connected to backtesting service backend"); Ok(Self { client, health_checker, backend_url: backend_url.to_string(), channel, }) } /// Perform background health check (should be called periodically) pub async fn background_health_check(&self) { self.health_checker .perform_health_check(&self.channel) .await; } /// Check if backend is currently healthy pub async fn is_backend_healthy(&self) -> bool { self.health_checker.is_healthy().await } /// Forward request with health checking and latency tracking async fn forward_with_health_check( &self, operation: &str, forward_fn: F, ) -> Result where F: FnOnce() -> Fut, Fut: std::future::Future>, { // Check health before forwarding if !self.health_checker.is_healthy().await { return Err(Status::unavailable(format!( "Backtesting service backend is unhealthy ({})", self.backend_url ))); } // Track latency let start = Instant::now(); // Forward request let result = forward_fn().await; let elapsed = start.elapsed(); // Record health outcome match &result { Ok(_) => { self.health_checker.record_success().await; debug!( operation = operation, latency_us = elapsed.as_micros(), "Forwarded request to backtesting service" ); }, Err(status) => { self.health_checker.record_failure().await; error!( operation = operation, error = %status, latency_us = elapsed.as_micros(), "Failed to forward request to backtesting service" ); }, } result } } #[tonic::async_trait] impl BacktestingService for BacktestingServiceProxy { // Define streaming response type for subscribe_backtest_progress type SubscribeBacktestProgressStream = tonic::codec::Streaming; /// Forward start_backtest request to backend #[instrument(skip_all)] async fn start_backtest( &self, request: Request, ) -> Result, Status> { self.forward_with_health_check("start_backtest", || async { // Clone the client (cheap operation, reuses connection) let mut client = self.client.clone(); // Extract inner request and forward let inner_request = request.into_inner(); // Forward to backend - zero additional allocations client.start_backtest(inner_request).await }) .await } /// Forward get_backtest_status request to backend #[instrument(skip_all)] async fn get_backtest_status( &self, request: Request, ) -> Result, Status> { self.forward_with_health_check("get_backtest_status", || async { let mut client = self.client.clone(); let inner_request = request.into_inner(); client.get_backtest_status(inner_request).await }) .await } /// Forward get_backtest_results request to backend #[instrument(skip_all)] async fn get_backtest_results( &self, request: Request, ) -> Result, Status> { self.forward_with_health_check("get_backtest_results", || async { let mut client = self.client.clone(); let inner_request = request.into_inner(); client.get_backtest_results(inner_request).await }) .await } /// Forward list_backtests request to backend #[instrument(skip_all)] async fn list_backtests( &self, request: Request, ) -> Result, Status> { self.forward_with_health_check("list_backtests", || async { let mut client = self.client.clone(); let inner_request = request.into_inner(); client.list_backtests(inner_request).await }) .await } /// Forward subscribe_backtest_progress streaming request to backend #[instrument(skip_all)] async fn subscribe_backtest_progress( &self, request: Request, ) -> Result, Status> { self.forward_with_health_check("subscribe_backtest_progress", || async { let mut client = self.client.clone(); let inner_request = request.into_inner(); // Forward streaming request - the response is already a stream let response = client.subscribe_backtest_progress(inner_request).await?; // Extract the stream and pass it through Ok(Response::new(response.into_inner())) }) .await } /// Forward stop_backtest request to backend #[instrument(skip_all)] async fn stop_backtest( &self, request: Request, ) -> Result, Status> { self.forward_with_health_check("stop_backtest", || async { let mut client = self.client.clone(); let inner_request = request.into_inner(); client.stop_backtest(inner_request).await }) .await } } // Implement BacktestingService for Arc to enable Arc wrapping #[tonic::async_trait] impl BacktestingService for Arc { type SubscribeBacktestProgressStream = tonic::codec::Streaming; #[instrument(skip_all)] async fn start_backtest( &self, request: Request, ) -> Result, Status> { self.as_ref().start_backtest(request).await } #[instrument(skip_all)] async fn get_backtest_status( &self, request: Request, ) -> Result, Status> { self.as_ref().get_backtest_status(request).await } #[instrument(skip_all)] async fn get_backtest_results( &self, request: Request, ) -> Result, Status> { self.as_ref().get_backtest_results(request).await } #[instrument(skip_all)] async fn list_backtests( &self, request: Request, ) -> Result, Status> { self.as_ref().list_backtests(request).await } #[instrument(skip_all)] async fn subscribe_backtest_progress( &self, request: Request, ) -> Result, Status> { self.as_ref().subscribe_backtest_progress(request).await } #[instrument(skip_all)] async fn stop_backtest( &self, request: Request, ) -> Result, Status> { self.as_ref().stop_backtest(request).await } } #[cfg(test)] mod tests { use super::*; #[tokio::test] async fn test_health_checker_success() { let checker = HealthChecker::new(3, Duration::from_secs(10)); assert_eq!(checker.get_state().await, HealthState::Healthy); assert!(checker.is_healthy().await); checker.record_success().await; assert_eq!(checker.get_state().await, HealthState::Healthy); } #[tokio::test] async fn test_health_checker_failure() { let checker = HealthChecker::new(3, Duration::from_secs(10)); checker.record_failure().await; assert_eq!(checker.get_state().await, HealthState::Degraded); checker.record_failure().await; assert_eq!(checker.get_state().await, HealthState::Degraded); checker.record_failure().await; assert_eq!(checker.get_state().await, HealthState::Unhealthy); assert!(!checker.is_healthy().await); } #[tokio::test] async fn test_health_checker_recovery() { let checker = HealthChecker::new(3, Duration::from_secs(10)); // Mark as unhealthy for _ in 0..3 { checker.record_failure().await; } assert_eq!(checker.get_state().await, HealthState::Unhealthy); // Recovery checker.record_success().await; assert_eq!(checker.get_state().await, HealthState::Healthy); assert!(checker.is_healthy().await); } }