diff --git a/services/broker_gateway_service/src/service.rs b/services/broker_gateway_service/src/service.rs index d75a98740..521be1d24 100644 --- a/services/broker_gateway_service/src/service.rs +++ b/services/broker_gateway_service/src/service.rs @@ -277,20 +277,15 @@ impl broker_gateway_service_server::BrokerGatewayService for BrokerGatewayServic let req = request.into_inner(); info!("GetAccountState called: account_id={}", req.account_id); - // MVP: Return placeholder data (no actual broker query) - let response = GetAccountStateResponse { - account_id: req.account_id, - cash_balance: 100_000.0, - equity: 100_000.0, - margin_used: 0.0, - margin_available: 100_000.0, - buying_power: 400_000.0, // 4x leverage - unrealized_pnl: 0.0, - realized_pnl: 0.0, - last_updated: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), - }; - - Ok(Response::new(response)) + // Broker account query not yet implemented — return explicit error + // rather than silently returning hardcoded fake data + warn!( + account_id = %req.account_id, + "get_account_state called but broker query is not implemented — returning error" + ); + Err(Status::failed_precondition( + "Account state query not yet implemented — broker connection MVP does not support live account queries", + )) } #[instrument(skip(self), fields(account_id))] @@ -304,16 +299,16 @@ impl broker_gateway_service_server::BrokerGatewayService for BrokerGatewayServic req.account_id, req.symbol ); - // MVP: Return empty positions (no actual broker query) - let response = GetPositionsResponse { - positions: vec![], - total_equity: 100_000.0, - total_exposure: 0.0, - leverage_ratio: 0.0, - timestamp: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), - }; - - Ok(Response::new(response)) + // Broker position query not yet implemented — return explicit error + // rather than silently returning empty positions with fake equity + warn!( + account_id = %req.account_id, + symbol = ?req.symbol, + "get_positions called but broker query is not implemented — returning error" + ); + Err(Status::failed_precondition( + "Position query not yet implemented — broker connection MVP does not support live position queries", + )) } #[instrument(skip(self))] diff --git a/services/data_acquisition_service/src/service.rs b/services/data_acquisition_service/src/service.rs index 045133617..a6b52ba3c 100644 --- a/services/data_acquisition_service/src/service.rs +++ b/services/data_acquisition_service/src/service.rs @@ -114,7 +114,53 @@ impl DataAcquisitionService for DataAcquisitionServiceImpl { store.insert(job_id.to_string(), job_details.clone()); } - // TODO: Start background download task + // Spawn background download task + { + let job_store = Arc::clone(&self.job_store); + let downloader = Arc::clone(&self.downloader); + let job_id_str = job_id.to_string(); + let download_job = crate::downloader::DownloadJob { + job_id, + dataset: req.dataset.clone(), + symbols: req.symbols.clone(), + start_date: req.start_date.clone(), + end_date: req.end_date.clone(), + schema: req.schema.clone(), + priority: req.priority as i32, + }; + + tokio::spawn(async move { + // Mark as downloading + { + let mut store = job_store.write().await; + if let Some(job) = store.get_mut(&job_id_str) { + job.status = DownloadStatus::Downloading as i32; + job.started_at = chrono::Utc::now().timestamp(); + } + } + + match downloader.download(&download_job).await { + Ok(path) => { + let mut store = job_store.write().await; + if let Some(job) = store.get_mut(&job_id_str) { + job.status = DownloadStatus::Completed as i32; + job.progress_percentage = 100.0; + job.completed_at = chrono::Utc::now().timestamp(); + job.local_path = path.to_string_lossy().to_string(); + } + } + Err(e) => { + let mut store = job_store.write().await; + if let Some(job) = store.get_mut(&job_id_str) { + job.status = DownloadStatus::Failed as i32; + job.completed_at = chrono::Utc::now().timestamp(); + job.error_message = e.to_string(); + job.retry_count += 1; + } + } + } + }); + } Ok(Response::new(ScheduleDownloadResponse { job_id: job_id.to_string(), @@ -147,7 +193,6 @@ impl DataAcquisitionService for DataAcquisitionServiceImpl { ) -> Result, Status> { let req = request.into_inner(); - // TODO: Implement actual cancellation logic let job_id = Uuid::parse_str(&req.job_id) .map_err(|e| Status::invalid_argument(format!("Invalid job ID: {}", e)))?; @@ -232,11 +277,38 @@ impl DataAcquisitionService for DataAcquisitionServiceImpl { let mut details = std::collections::HashMap::new(); details.insert("service".to_string(), "data_acquisition".to_string()); details.insert("version".to_string(), env!("CARGO_PKG_VERSION").to_string()); + + // Check job store accessibility + let store = self.job_store.read().await; + let total_jobs = store.len(); + let active_jobs = store + .values() + .filter(|j| j.status == DownloadStatus::Downloading as i32) + .count(); + let failed_jobs = store + .values() + .filter(|j| j.status == DownloadStatus::Failed as i32) + .count(); + drop(store); + + // Check downloader active count + let active_downloads = self.downloader.active_download_count().await; + + details.insert("total_jobs".to_string(), total_jobs.to_string()); + details.insert("active_jobs".to_string(), active_jobs.to_string()); + details.insert("failed_jobs".to_string(), failed_jobs.to_string()); + details.insert( + "active_downloads".to_string(), + active_downloads.to_string(), + ); details.insert("status".to_string(), "operational".to_string()); Ok(Response::new(HealthCheckResponse { healthy: true, - message: "Service is healthy".to_string(), + message: format!( + "Service is healthy: {} active downloads, {} total jobs", + active_downloads, total_jobs + ), details, })) } diff --git a/services/ml_training_service/src/health.rs b/services/ml_training_service/src/health.rs index d234c1796..e8f412742 100644 --- a/services/ml_training_service/src/health.rs +++ b/services/ml_training_service/src/health.rs @@ -1,11 +1,28 @@ -//! Health check endpoints for backtesting service +//! Health check endpoints for ML training service //! //! Provides HTTP-based health and readiness endpoints that don't require TLS, -//! allowing Docker Compose health checks to work without client certificates. +//! allowing Docker Compose / Kubernetes health checks to work without client certificates. +//! +//! The readiness probe validates that the training orchestrator has fully initialized +//! before Kubernetes routes traffic to this instance. -use axum::{routing::get, Json, Router}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; + +use axum::{extract::State, routing::get, Json, Router}; use serde::Serialize; +/// Shared health state for readiness probing. +/// +/// The `ready` flag is set to `true` once the training orchestrator has started +/// successfully. Until then, the `/ready` endpoint returns `"not_ready"` so that +/// Kubernetes does not route traffic to an instance that cannot serve requests. +#[derive(Clone)] +pub struct HealthState { + /// Whether the service is ready to accept requests. + pub ready: Arc, +} + /// Health status response #[derive(Serialize)] pub struct HealthResponse { @@ -18,13 +35,18 @@ pub struct HealthResponse { } /// Creates the health check router with /health and /ready endpoints -pub fn health_router() -> Router { +pub fn health_router(state: HealthState) -> Router { Router::new() .route("/health", get(health_check)) .route("/ready", get(readiness_check)) + .with_state(state) } -/// Liveness probe - indicates service is running +/// Liveness probe - indicates service process is running. +/// +/// This intentionally does NOT check any downstream dependencies. If this +/// handler can execute, the process is alive — which is exactly what a +/// Kubernetes liveness probe should verify. async fn health_check() -> Json { Json(HealthResponse { status: "healthy", @@ -33,12 +55,18 @@ async fn health_check() -> Json { }) } -/// Readiness probe - indicates service is ready to accept requests -async fn readiness_check() -> Json { - // In the future, can add database connection checks here - // For now, if the service is running, it's ready +/// Readiness probe - indicates service is ready to accept requests. +/// +/// Returns `"ready"` only when the training orchestrator has fully initialized. +/// Before that, returns `"not_ready"` so Kubernetes will not route traffic here. +async fn readiness_check(State(state): State) -> Json { + let status = if state.ready.load(Ordering::Relaxed) { + "ready" + } else { + "not_ready" + }; Json(HealthResponse { - status: "ready", + status, service: "ml_training", version: env!("CARGO_PKG_VERSION"), }) diff --git a/services/ml_training_service/src/main.rs b/services/ml_training_service/src/main.rs index 571777ad2..ae50715c4 100644 --- a/services/ml_training_service/src/main.rs +++ b/services/ml_training_service/src/main.rs @@ -6,6 +6,7 @@ #![deny(clippy::unwrap_used, clippy::expect_used)] use std::net::SocketAddr; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -310,6 +311,9 @@ async fn serve(args: ServeArgs) -> Result<()> { info!("Storage backend initialized with ConfigManager integration"); + // Create health state — readiness flag starts as false + let health_ready = Arc::new(AtomicBool::new(false)); + // Initialize orchestrator let mut orchestrator = TrainingOrchestrator::new( ml_config.clone(), @@ -326,6 +330,8 @@ async fn serve(args: ServeArgs) -> Result<()> { .context("Failed to start orchestrator")?; let orchestrator = Arc::new(orchestrator); + // Mark service as ready now that the orchestrator is running + health_ready.store(true, Ordering::Relaxed); info!("Training orchestrator started"); // Initialize TLS configuration for mTLS @@ -481,7 +487,10 @@ async fn serve(args: ServeArgs) -> Result<()> { info!("Starting health check server on {}", health_addr); - let health_app = health::health_router(); + let health_state = health::HealthState { + ready: Arc::clone(&health_ready), + }; + let health_app = health::health_router(health_state); tokio::spawn(async move { let health_listener = match tokio::net::TcpListener::bind(health_addr).await {