#![warn(missing_docs)] #![deny(clippy::unwrap_used)] #![deny(clippy::expect_used)] //! # Foxhunt Backtesting Service //! //! Standalone backtesting service for the Foxhunt HFT trading system. //! This service provides comprehensive strategy testing and performance analysis capabilities. use anyhow::{Context, Result}; use rustls::crypto::CryptoProvider; use std::net::SocketAddr; use tonic::transport::Server; use tracing::{error, info, warn}; use common::metrics::GrpcMetricsLayer; mod health; use backtesting_service::foxhunt; use backtesting_service::repository_impl::create_repositories; use backtesting_service::service::BacktestingServiceImpl; use backtesting_service::storage::StorageManager; use backtesting_service::tls_config::BacktestingServiceTlsConfig; use config::structures::BacktestingDatabaseConfig; use std::sync::Arc; /// Main entry point for the backtesting service #[tokio::main] async fn main() -> Result<()> { // Wave 77 Agent 3: Initialize crypto provider FIRST before any TLS operations // This fixes the "Could not automatically determine the process-level CryptoProvider" panic CryptoProvider::install_default(rustls::crypto::ring::default_provider()) .map_err(|_| anyhow::anyhow!("Failed to install default crypto provider"))?; // Initialize observability (JSON logging + OpenTelemetry tracing via OTLP) let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") .unwrap_or_else(|_| "http://localhost:4317".to_string()); if let Err(e) = common::observability::init_observability( "backtesting_service", Some(&otlp_endpoint), ) { eprintln!("Failed to initialize observability: {}", e); } info!("Starting Foxhunt Backtesting Service"); // Get database configuration from environment let database_url = std::env::var("DATABASE_URL").context("DATABASE_URL environment variable is required")?; // Wave 67 Agent 2: Optimized database configuration for backtesting workloads // Based on Wave 66 Agent 8 analysis - increased statement cache for better performance let database_config = BacktestingDatabaseConfig { database_url, max_connections: Some(10), min_connections: Some(2), acquire_timeout_ms: Some(5000), statement_cache_capacity: Some(500), // Increased from 100 to 500 for better cache hit rate enable_logging: Some(false), }; // Configuration loaded from environment info!("Configuration loaded from environment variables"); info!("Backtesting configuration loaded successfully"); // Initialize storage manager let storage_manager = Arc::new( StorageManager::new(&database_config) .await .context("Failed to initialize storage manager")?, ); // Create repositories with dependency injection let repositories = Arc::new( create_repositories(storage_manager) .await .context("Failed to create repositories")?, ); // Initialize the service with repository injection let service = BacktestingServiceImpl::new(repositories) .await .context("Failed to initialize backtesting service")?; // Initialize TLS configuration for mTLS // Read TLS configuration from environment variables (Docker deployment) let tls_enabled = std::env::var("TLS_ENABLED") .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(false); info!("TLS Configuration:"); info!(" TLS Enabled: {}", tls_enabled); let tls_config = if tls_enabled { let cert_path = std::env::var("TLS_CERT_PATH") .context("TLS_CERT_PATH environment variable required when TLS is enabled")?; let key_path = std::env::var("TLS_KEY_PATH") .context("TLS_KEY_PATH environment variable required when TLS is enabled")?; let ca_cert_path = std::env::var("TLS_CA_PATH") .context("TLS_CA_PATH environment variable required for mTLS")?; let require_client_cert = std::env::var("TLS_REQUIRE_CLIENT_CERT") .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(true); // Default to requiring client certs for security info!(" Certificate Path: {}", cert_path); info!(" Key Path: {}", key_path); info!(" CA Cert Path: {}", ca_cert_path); info!(" Require Client Cert: {}", require_client_cert); BacktestingServiceTlsConfig::from_files( &cert_path, &key_path, &ca_cert_path, require_client_cert, ) .await .context("Failed to initialize TLS configuration from files")? } else { info!(" TLS is disabled - using default configuration"); info!(" ⚠️ WARNING: Running without TLS in production is NOT recommended"); // Create a default config that won't be used (server won't apply TLS) BacktestingServiceTlsConfig::from_files( "/tmp/foxhunt/certs/server-cert.pem", "/tmp/foxhunt/certs/server-key.pem", "/tmp/foxhunt/certs/ca/ca-cert.pem", false, ) .await .unwrap_or_else(|e| { warn!( "Failed to load default TLS config (expected when TLS disabled): {}", e ); // Return a minimal config that won't be used BacktestingServiceTlsConfig { server_identity: tonic::transport::Identity::from_pem(vec![0], vec![0]), ca_certificate: tonic::transport::Certificate::from_pem(vec![0]), require_client_cert: false, } }) }; // Setup gRPC server let grpc_port = std::env::var("GRPC_PORT") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(50053); let addr: SocketAddr = format!("0.0.0.0:{}", grpc_port) .parse() .context("Invalid server address in configuration")?; info!("Starting gRPC server on {}", addr); // Wave 67 Agent 3: HTTP/2 streaming performance optimizations // Apply same optimizations as other services for consistency let enable_http2_opts = std::env::var("ENABLE_HTTP2_OPTIMIZATIONS") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(true); if enable_http2_opts { info!("✅ HTTP/2 optimizations enabled:"); info!(" - tcp_nodelay: true (-40ms Nagle delay)"); info!(" - Stream window: 1MB"); info!(" - Connection window: 10MB"); info!(" - Adaptive window: true"); info!(" - Max streams: 10,000 (production scale)"); } else { info!("⚠️ HTTP/2 optimizations disabled via feature flag"); } // Start the server with HTTP/2 optimizations let mut server_builder = Server::builder() .layer(GrpcMetricsLayer::new("backtesting-service")); if enable_http2_opts { server_builder = server_builder .tcp_nodelay(true) // Critical: eliminates 40ms Nagle delay .http2_keepalive_interval(Some(std::time::Duration::from_secs(30))) .http2_keepalive_timeout(Some(std::time::Duration::from_secs(10))) .initial_stream_window_size(Some(1024 * 1024)) // 1MB .initial_connection_window_size(Some(10 * 1024 * 1024)) // 10MB .http2_adaptive_window(Some(true)) .max_concurrent_streams(Some(10_000)); // Increased from 1,024 to 10,000 for production scale } // Initialize Prometheus metrics backtesting_service::simple_metrics::init_metrics(); let service_start = std::time::Instant::now(); // Spawn uptime updater tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); loop { interval.tick().await; backtesting_service::simple_metrics::update_uptime(service_start); } }); // Start Prometheus metrics HTTP endpoint on port 9093 tokio::spawn(async { use axum::{routing::get, Router}; use prometheus::{Encoder, TextEncoder}; async fn metrics_handler() -> Result { let encoder = TextEncoder::new(); let metric_families = prometheus::gather(); let mut buffer = vec![]; encoder .encode(&metric_families, &mut buffer) .map_err(|e| format!("Failed to encode metrics: {}", e))?; String::from_utf8(buffer) .map_err(|e| format!("Failed to convert metrics to UTF-8: {}", e)) } async fn metrics_handler_wrapper() -> String { metrics_handler() .await .unwrap_or_else(|e| format!("Error: {}", e)) } let metrics_app = Router::new().route("/metrics", get(metrics_handler_wrapper)); let metrics_addr = "0.0.0.0:9093"; info!( "Prometheus metrics endpoint listening on http://{}", metrics_addr ); let listener = match tokio::net::TcpListener::bind(metrics_addr).await { Ok(l) => l, Err(e) => { error!("Failed to bind metrics endpoint {}: {}", metrics_addr, e); return; }, }; if let Err(e) = axum::serve(listener, metrics_app).await { error!("Metrics server failed: {}", e); } }); // Start HTTP health check server on port 8082 (no TLS required) let health_port = std::env::var("HEALTH_PORT") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(8082); let health_addr: SocketAddr = format!("0.0.0.0:{}", health_port) .parse() .context("Invalid health server address")?; info!("Starting health check server on {}", health_addr); let health_app = health::health_router(); tokio::spawn(async move { let health_listener = match tokio::net::TcpListener::bind(health_addr).await { Ok(l) => l, Err(e) => { error!("Failed to bind health server {}: {}", health_addr, e); return; }, }; if let Err(e) = axum::serve(health_listener, health_app).await { error!("Health server failed: {}", e); } }); // Setup gRPC health reporting service let (health_reporter, health_service) = tonic_health::server::health_reporter(); // Mark backtesting service as SERVING health_reporter .set_serving::>() .await; info!("gRPC health service configured - backtesting service marked as SERVING"); // Build server with optional TLS and HTTP/2 optimizations let router = if tls_enabled { info!("✅ TLS enabled - configuring mTLS for gRPC server"); server_builder .tls_config(tls_config.to_server_tls_config())? .add_service(health_service) // Add gRPC health service first .add_service( foxhunt::tli::backtesting_service_server::BacktestingServiceServer::new(service), ) } else { info!("⚠️ TLS disabled - running gRPC server without encryption"); server_builder .add_service(health_service) // Add gRPC health service first .add_service( foxhunt::tli::backtesting_service_server::BacktestingServiceServer::new(service), ) }; info!( "🚀 Backtesting Service ready - starting gRPC server on {}", addr ); router .serve_with_shutdown(addr, shutdown_signal()) .await .context("gRPC server failed")?; info!("Backtesting Service stopped gracefully"); Ok(()) } /// Waits for a CTRL+C signal for graceful shutdown async fn shutdown_signal() { if let Err(e) = tokio::signal::ctrl_c().await { tracing::error!("Failed to install CTRL+C handler: {}", e); } tracing::info!("Shutdown signal received, stopping gracefully..."); }