gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'worktree-pods-versioning'

Browse Source 
# Conflicts:
#	bin/fxt/src/error.rs
This commit is contained in:
jgrusewski
2026-03-04 14:59:55 +01:00
parent 091dee8dc5 e3c8428db4
commit ca06f71c88
33 changed files with 1934 additions and 3510 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
bin/fxt/benches/client_performance.rs
View File

@@ -284,7 +284,7 @@ fn bench_error_scenarios(c: &mut Criterion) {
// Connection error simulation
group.bench_function("connection_error", |b| {
b.iter(|| {
let error = TliError::Connection(black_box("Connection refused".to_string()));
let error = FxtError::Connection(black_box("Connection refused".to_string()));
let formatted = format!("{}", error);
black_box(formatted)
})
@@ -293,7 +293,7 @@ fn bench_error_scenarios(c: &mut Criterion) {
// Invalid request error
group.bench_function("invalid_request_error", |b| {
b.iter(|| {
let error = TliError::InvalidRequest(black_box("Invalid order quantity".to_string()));
let error = FxtError::InvalidRequest(black_box("Invalid order quantity".to_string()));
let formatted = format!("{}", error);
black_box(formatted)
})
@@ -303,7 +303,7 @@ fn bench_error_scenarios(c: &mut Criterion) {
group.bench_function("not_connected_error", |b| {
b.iter(|| {
let error =
TliError::Connection(black_box("Trading service not connected".to_string()));
FxtError::Connection(black_box("Trading service not connected".to_string()));
let formatted = format!("{}", error);
black_box(formatted)
})
@@ -314,8 +314,8 @@ fn bench_error_scenarios(c: &mut Criterion) {
b.iter(|| {
let io_error =
std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "Connection refused");
let tli_error: TliError = io_error.into();
let formatted = format!("{}", black_box(tli_error));
let fxt_error: FxtError = io_error.into();
let formatted = format!("{}", black_box(fxt_error));
black_box(formatted)
})
});

8
bin/fxt/benches/configuration_benchmarks.rs
View File

@@ -410,7 +410,7 @@ fn bench_error_handling(c: &mut Criterion) {
// Result creation and matching
group.bench_function("success_result", |b| {
b.iter(|| {
let result: TliResult<i32> = Ok(black_box(42));
let result: FxtResult<i32> = Ok(black_box(42));
match result {
Ok(value) => black_box(value),
Err(_) => 0,
@@ -420,7 +420,7 @@ fn bench_error_handling(c: &mut Criterion) {
group.bench_function("error_result", |b| {
b.iter(|| {
let result: TliResult<i32> = Err(TliError::InvalidRequest("test error".to_string()));
let result: FxtResult<i32> = Err(FxtError::InvalidRequest("test error".to_string()));
match result {
Ok(value) => value,
Err(_) => black_box(0),
@@ -431,14 +431,14 @@ fn bench_error_handling(c: &mut Criterion) {
// Error creation
group.bench_function("error_creation", |b| {
b.iter(|| {
let error = TliError::Connection(black_box("Connection failed".to_string()));
let error = FxtError::Connection(black_box("Connection failed".to_string()));
black_box(error)
})
});
// Error message formatting
group.bench_function("error_formatting", |b| {
let error = TliError::InvalidSymbol("TEST".to_string());
let error = FxtError::InvalidSymbol("TEST".to_string());
b.iter(|| {
let formatted = format!("{}", black_box(&error));
black_box(formatted)

4
bin/fxt/build.rs
View File

@@ -1,7 +1,7 @@
fn main() -> Result<(), Box<dyn std::error::Error>> {
// proto/ at workspace root — single source of truth for all services.
// Note: fxt_trading.proto (package foxhunt.tli) is excluded — it is the
// legacy fat-client proto kept only for API Gateway backward compatibility.
// legacy fat-client proto kept for wire compatibility (package name retained).
let proto_root = "../../proto";
let protos: Vec<String> = [
"trading",
@@ -14,7 +14,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"monitoring",
"risk",
"data_acquisition",
"config_service",
]
.iter()
.map(|p| format!("{proto_root}/{p}.proto"))
@@ -40,7 +39,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"monitoring",
"risk",
"data_acquisition",
"config_service",
] {
println!("cargo:rerun-if-changed={proto_root}/{proto}.proto");
}

2
bin/fxt/src/auth/jwt_generator.rs
View File

@@ -51,7 +51,7 @@ impl Default for JwtConfig {
let resolved_secret = std::env::var("JWT_SECRET")
.ok()
.or_else(|| {
crate::config::TliConfig::load()
crate::config::FxtConfig::load()
.ok()
.and_then(|c| c.jwt_secret)
})

95
bin/fxt/src/auth/token_manager.rs
View File

@@ -1,7 +1,7 @@
//! JWT token management with automatic refresh
//!
//! Manages access tokens and refresh tokens via OS keyring for secure persistence.
//! Since TLI is a CLI tool (not a long-running service), tokens are read from
//! Since FXT is a CLI tool (not a long-running service), tokens are read from
//! keyring on each access rather than cached in memory.
use anyhow::{Context, Result};
@@ -11,6 +11,35 @@ use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;
/// Recursively copy a directory tree (used for token storage migration fallback).
fn copy_dir_recursive(src: &std::path::Path, dst: &std::path::Path) -> Result<()> {
std::fs::create_dir_all(dst)
.with_context(|| format!("Failed to create directory {}", dst.display()))?;
for dir_entry in std::fs::read_dir(src)
.with_context(|| format!("Failed to read directory {}", src.display()))?
{
let entry = dir_entry.context("Failed to read directory entry")?;
let src_path = entry.path();
let dst_path = dst.join(entry.file_name());
if src_path.is_dir() {
copy_dir_recursive(&src_path, &dst_path)?;
} else {
std::fs::copy(&src_path, &dst_path).with_context(|| {
format!(
"Failed to copy {} to {}",
src_path.display(),
dst_path.display()
)
})?;
}
}
Ok(())
}
/// Buffer (in seconds) before actual JWT expiry at which we consider the token expired.
/// This ensures we refresh proactively rather than racing against clock skew.
const TOKEN_REFRESH_BUFFER_SECS: u64 = 60;
/// JWT token information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TokenInfo {
@@ -25,15 +54,14 @@ pub struct TokenInfo {
impl TokenInfo {
/// Check if the access token is expired or nearing expiration
///
/// Returns true if the token expires within the next 60 seconds
/// Returns true if the token expires within [`TOKEN_REFRESH_BUFFER_SECS`] seconds
pub fn is_expired(&self) -> bool {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or(Duration::from_secs(0))
.as_secs();
// Consider expired if within 60 seconds of expiration
self.expires_at <= now + 60
self.expires_at <= now + TOKEN_REFRESH_BUFFER_SECS
}
/// Get remaining time until token expiration
@@ -64,6 +92,7 @@ fn extract_token_expiry(token: &str) -> Result<u64> {
validation.validate_exp = false;
validation.validate_aud = false; // Disable audience validation for flexibility
// Key value is irrelevant — signature validation is disabled above
let token_data = decode::<JwtClaims>(token, &DecodingKey::from_secret(b"dummy"), &validation)
.context("Failed to decode JWT token")?;
@@ -103,7 +132,7 @@ impl KeyringTokenStorage {
/// Create a new keyring token storage
///
/// # Arguments
/// * `service_name` - Application service name (e.g., "foxhunt-tli")
/// * `service_name` - Application service name (e.g., "foxhunt-fxt")
///
/// * `username` - Username for keyring entry
pub const fn new(service_name: String, username: String) -> Self {
@@ -152,7 +181,7 @@ impl TokenStorage for KeyringTokenStorage {
let owned_token = token.to_owned();
tokio::task::spawn_blocking(move || {
let entry = keyring::Entry::new("foxhunt-tli-access", "default")
let entry = keyring::Entry::new("foxhunt-fxt-access", "default")
.context("Failed to create keyring entry for access token")?;
entry
@@ -168,7 +197,7 @@ impl TokenStorage for KeyringTokenStorage {
async fn get_access_token(&self) -> Result<Option<String>> {
tokio::task::spawn_blocking(move || {
let entry = keyring::Entry::new("foxhunt-tli-access", "default")
let entry = keyring::Entry::new("foxhunt-fxt-access", "default")
.context("Failed to create keyring entry for access token")?;
match entry.get_password() {
@@ -242,7 +271,7 @@ impl TokenStorage for KeyringTokenStorage {
async fn clear_access_token(&self) -> Result<()> {
tokio::task::spawn_blocking(move || {
let entry = keyring::Entry::new("foxhunt-tli-access", "default")
let entry = keyring::Entry::new("foxhunt-fxt-access", "default")
.context("Failed to create keyring entry for access token")?;
match entry.delete_credential() {
@@ -266,7 +295,7 @@ impl TokenStorage for KeyringTokenStorage {
/// object cannot be retrieved by a different Entry object.
///
/// Security notes:
/// - Files stored in `~/.config/foxhunt-tli/tokens/`
/// - Files stored in `~/.config/foxhunt-fxt/tokens/`
/// - Directory permissions: 700 (owner read/write/execute only)
/// - File permissions: 600 (owner read/write only)
/// - AES-256-GCM encryption provides production-grade security
@@ -281,10 +310,27 @@ impl FileTokenStorage {
///
/// Creates token directory with 700 permissions if it doesn't exist.
pub fn new() -> Result<Self> {
let token_dir = dirs::config_dir()
.context("Cannot determine config directory")?
.join("foxhunt-tli")
.join("tokens");
let config_base = dirs::config_dir().context("Cannot determine config directory")?;
let new_dir = config_base.join("foxhunt-fxt");
let old_dir = config_base.join("foxhunt-tli");
// Migrate from legacy foxhunt-tli directory if it exists and foxhunt-fxt does not
if !new_dir.exists() && old_dir.exists() {
tracing::info!(
"Migrating token storage from {} to {}",
old_dir.display(),
new_dir.display()
);
if let Err(e) = std::fs::rename(&old_dir, &new_dir) {
// rename() fails across mount points; fall back to copy + remove
tracing::warn!("rename() failed ({}), falling back to copy", e);
copy_dir_recursive(&old_dir, &new_dir)?;
// Best-effort cleanup of old dir; failure is non-fatal
drop(std::fs::remove_dir_all(&old_dir));
}
}
let token_dir = new_dir.join("tokens");
// Create directory with 700 permissions (owner only)
std::fs::create_dir_all(&token_dir).context("Failed to create token directory")?;
@@ -410,9 +456,24 @@ impl FileTokenStorage {
impl Default for FileTokenStorage {
fn default() -> Self {
// FileTokenStorage::new() creates dirs on the filesystem; Default can't return Result
#[allow(clippy::expect_used)]
Self::new().expect("Failed to create FileTokenStorage")
// Try the normal path first; fall back to /tmp/foxhunt-fxt/tokens/ if it fails.
// This avoids a panic when the user's config directory is unavailable.
Self::new().unwrap_or_else(|e| {
tracing::warn!(
"FileTokenStorage::new() failed ({}), falling back to /tmp/foxhunt-fxt/tokens/",
e
);
let fallback = std::path::PathBuf::from("/tmp/foxhunt-fxt/tokens");
Self::with_directory(fallback).unwrap_or_else(|e2| {
// Last resort: construct without creating dirs -- will fail at read/write
// time rather than panicking here.
tracing::error!("Fallback token storage also failed: {}", e2);
Self {
token_dir: std::path::PathBuf::from("/tmp/foxhunt-fxt/tokens"),
key_manager: std::sync::Mutex::new(crate::auth::key_manager::KeyManager::new()),
}
})
})
}
}
@@ -561,7 +622,7 @@ impl TokenStorage for InMemoryTokenStorage {
/// Authentication token manager
///
/// Manages JWT tokens via keyring storage. Since TLI is a CLI tool,
/// Manages JWT tokens via keyring storage. Since FXT is a CLI tool,
/// tokens are read from keyring on each access rather than cached in memory.
pub struct AuthTokenManager<S: TokenStorage> {
/// Token storage backend

2
bin/fxt/src/commands/config.rs
View File

@@ -250,7 +250,7 @@ impl ConfigCommand {
}
ConfigAction::Env => {
let config = crate::config::TliConfig::load().unwrap_or_default();
let config = crate::config::FxtConfig::load().unwrap_or_default();
let config_path = dirs::home_dir()
.map(|h| h.join(".foxhunt").join("config.toml"))
.map(|p| p.display().to_string())

4
bin/fxt/src/commands/mcp.rs
View File

@@ -6,6 +6,7 @@
use anyhow::Result;
use clap::Parser;
use crate::grpc::FoxhuntClient;
use crate::mcp::server::McpServer;
/// Start MCP server (JSON-RPC 2.0 on stdin/stdout)
@@ -24,7 +25,8 @@ impl McpCommand {
/// Start the MCP server that exposes Foxhunt operations as tools
/// for LLM agents (e.g. Claude, Cursor).
pub async fn execute(&self) -> Result<()> {
let server = McpServer::new(self.api_url.clone());
let client = FoxhuntClient::connect(&self.api_url).await?;
let server = McpServer::new(self.api_url.clone(), client);
server.run().await
}
}

29
bin/fxt/src/config.rs
View File

@@ -39,6 +39,7 @@ fn default_ibkr_host() -> String {
fn default_ibkr_port() -> u16 {
4004
}
// Default IBKR client ID (99 = non-production default)
fn default_ibkr_client_id() -> i32 {
99
}
@@ -83,7 +84,7 @@ impl Default for BrokerConfig {
/// Settings can be loaded from `~/.foxhunt/config.toml` and overridden
/// by CLI arguments or environment variables.
#[derive(Debug, Serialize, Deserialize)]
pub struct TliConfig {
pub struct FxtConfig {
/// API Gateway URL (default: <http://localhost:50051>)
#[serde(default = "default_api_gateway_url")]
pub api_gateway_url: String,
@@ -117,7 +118,7 @@ fn default_token_storage() -> String {
"keyring".to_owned()
}
impl Default for TliConfig {
impl Default for FxtConfig {
fn default() -> Self {
Self {
api_gateway_url: default_api_gateway_url(),
@@ -129,7 +130,7 @@ impl Default for TliConfig {
}
}
impl TliConfig {
impl FxtConfig {
/// Load config from ~/.foxhunt/config.toml
///
/// Returns default configuration if file doesn't exist.
@@ -137,9 +138,9 @@ impl TliConfig {
///
/// # Examples
/// ```no_run
/// use fxt::config::TliConfig;
/// use fxt::config::FxtConfig;
///
/// let config = TliConfig::load().unwrap();
/// let config = FxtConfig::load().unwrap();
/// println!("API Gateway: {}", config.api_gateway_url);
/// ```
pub fn load() -> Result<Self> {
@@ -164,9 +165,9 @@ impl TliConfig {
///
/// # Examples
/// ```no_run
/// use fxt::config::TliConfig;
/// use fxt::config::FxtConfig;
///
/// let config = TliConfig {
/// let config = FxtConfig {
/// api_gateway_url: "http://localhost:50051".to_string(),
/// log_level: "debug".to_string(),
/// token_storage: "keyring".to_string(),
@@ -208,7 +209,7 @@ mod tests {
#[test]
fn test_default_config() {
// Test that Default trait provides correct values
let config = TliConfig::default();
let config = FxtConfig::default();
assert_eq!(config.api_gateway_url, "https://api.fxhnt.ai");
assert_eq!(config.log_level, "info");
assert_eq!(config.token_storage, "keyring");
@@ -216,7 +217,7 @@ mod tests {
#[test]
fn test_config_serialization() {
let config = TliConfig {
let config = FxtConfig {
api_gateway_url: "http://example.com:50051".to_owned(),
log_level: "debug".to_owned(),
token_storage: "file".to_owned(),
@@ -225,7 +226,7 @@ mod tests {
};
let toml_str = toml::to_string(&config).unwrap();
let parsed: TliConfig = toml::from_str(&toml_str).unwrap();
let parsed: FxtConfig = toml::from_str(&toml_str).unwrap();
assert_eq!(config.api_gateway_url, parsed.api_gateway_url);
assert_eq!(config.log_level, parsed.log_level);
@@ -236,7 +237,7 @@ mod tests {
fn test_load_config_succeeds() {
// load() should always succeed: returns file contents if present, defaults otherwise.
// Default-value assertions live in test_serde_defaults and test_default_config.
let result = TliConfig::load();
let result = FxtConfig::load();
assert!(result.is_ok());
}
@@ -244,7 +245,7 @@ mod tests {
fn test_serde_defaults() {
// Test that empty TOML string deserializes to defaults
let empty_toml = "";
let config: TliConfig = toml::from_str(empty_toml).unwrap();
let config: FxtConfig = toml::from_str(empty_toml).unwrap();
assert_eq!(config.api_gateway_url, "https://api.fxhnt.ai");
assert_eq!(config.log_level, "info");
assert_eq!(config.token_storage, "keyring");
@@ -263,7 +264,7 @@ mod tests {
port = 4002
client_id = 42
"#;
let config: TliConfig = toml::from_str(toml_str).unwrap();
let config: FxtConfig = toml::from_str(toml_str).unwrap();
assert_eq!(config.broker.gateway_url, "http://broker-gw:50060");
assert_eq!(config.broker.ibkr.host, "10.0.0.5");
assert_eq!(config.broker.ibkr.port, 4002);
@@ -272,7 +273,7 @@ mod tests {
#[test]
fn test_config_broker_defaults() {
let config: TliConfig = toml::from_str("").unwrap();
let config: FxtConfig = toml::from_str("").unwrap();
assert_eq!(config.broker.gateway_url, "http://localhost:50056");
assert_eq!(config.broker.ibkr.host, "127.0.0.1");
assert_eq!(config.broker.ibkr.port, 4004);

6
bin/fxt/src/error.rs
View File

@@ -7,7 +7,7 @@ use thiserror::Error;
/// Represents all possible errors that can occur in the FXT client system,
/// including network connectivity, configuration, validation, and protocol errors.
#[derive(Error, Debug)]
pub enum TliError {
pub enum FxtError {
/// Network connection or gRPC transport errors
///
/// Occurs when unable to establish or maintain connections to trading services,
@@ -81,7 +81,7 @@ pub enum TliError {
/// Result type alias for FXT operations
///
/// Standard Result type using `TliError` as the error variant.
/// Standard Result type using `FxtError` as the error variant.
///
/// Used throughout the FXT codebase for consistent error handling.
pub type TliResult<T> = Result<T, TliError>;
pub type FxtResult<T> = Result<T, FxtError>;

13
bin/fxt/src/grpc.rs
View File

@@ -176,17 +176,4 @@ impl FoxhuntClient {
)
}
/// API Gateway configuration service client (package `foxhunt.config`).
///
/// Distinct from [`Self::config`] which targets the trading service's
/// configuration proto (package `config`).
pub fn config_service(
&self,
) -> crate::proto::config_service::configuration_service_client::ConfigurationServiceClient<AuthChannel>
{
crate::proto::config_service::configuration_service_client::ConfigurationServiceClient::with_interceptor(
self.channel.clone(),
self.interceptor.clone(),
)
}
}

8
bin/fxt/src/lib.rs
View File

@@ -122,12 +122,4 @@ pub mod proto {
tonic::include_proto!("data_acquisition");
}
/// Config service protobuf definitions (package `foxhunt.config`)
///
/// Note: This is from `config_service.proto` (API Gateway config service),
/// distinct from `config.proto` (trading service config, package `config`).
#[allow(clippy::shadow_reuse, clippy::impl_trait_in_params, clippy::empty_structs_with_brackets)]
pub mod config_service {
tonic::include_proto!("foxhunt.config");
}
}

1635
bin/fxt/src/mcp/server.rs
View File

File diff suppressed because it is too large Load Diff

16
bin/fxt/src/tui/cockpits/services.rs
View File

@@ -44,7 +44,17 @@ fn render_service_grid(frame: &mut Frame, area: Rect, state: &AppState) {
.services
.iter()
.map(|s| {
let status_text = if s.healthy { "HEALTHY" } else { "DOWN" };
let status_text = if s.healthy {
"HEALTHY".to_owned()
} else {
match &s.error {
Some(msg) if !msg.is_empty() => {
let truncated: String = msg.chars().take(20).collect();
format!("DOWN: {truncated}")
}
_ => "DOWN".to_owned(),
}
};
let status_style = theme::status_style(s.healthy);
let latency_style = if s.latency_ms > 500.0 {
@@ -63,7 +73,7 @@ fn render_service_grid(frame: &mut Frame, area: Rect, state: &AppState) {
Row::new(vec![
Span::styled(s.name.clone(), Style::default().fg(theme::TEXT)),
Span::styled((*status_text).to_owned(), status_style),
Span::styled(status_text, status_style),
Span::styled(latency_text, latency_style),
Span::styled(s.uptime.clone(), theme::muted_style()),
Span::styled(s.version.clone(), theme::muted_style()),
@@ -75,7 +85,7 @@ fn render_service_grid(frame: &mut Frame, area: Rect, state: &AppState) {
rows,
[
Constraint::Length(20),
Constraint::Length(10),
Constraint::Length(28),
Constraint::Length(10),
Constraint::Length(10),
Constraint::Length(10),

19
bin/fxt/src/tui/cockpits/training.rs
View File

@@ -51,7 +51,7 @@ fn render_sessions_table(frame: &mut Frame, area: Rect, state: &AppState) {
Row::new(vec![
s.model.clone(),
format!("{}", s.fold),
format!("{}/{}", s.epoch, s.max_epochs),
s.epoch.to_string(),
format!("{:.4}", s.loss),
format!("{:.4}", s.val_loss),
format!("{:.1}", s.batches_per_sec),
@@ -61,6 +61,11 @@ fn render_sessions_table(frame: &mut Frame, area: Rect, state: &AppState) {
})
.collect();
let block_title = if state.active_jobs > 0 {
format!(" Active Training Sessions -- {} K8s jobs ", state.active_jobs)
} else {
" Active Training Sessions ".to_owned()
};
let table = Table::new(
rows,
[
@@ -74,7 +79,7 @@ fn render_sessions_table(frame: &mut Frame, area: Rect, state: &AppState) {
],
)
.header(header)
.block(theme::block(" Active Training Sessions "));
.block(theme::block(&block_title));
frame.render_widget(table, area);
}
@@ -139,16 +144,18 @@ fn render_gpu_panel(frame: &mut Frame, area: Rect, state: &AppState) {
theme::SUCCESS
};
let power_text = if state.gpu.power_limit_watts > 0 {
format!("{}W / {}W", state.gpu.power_watts, state.gpu.power_limit_watts)
} else {
format!("{}W", state.gpu.power_watts)
};
let info_items = vec![ListItem::new(Line::from(vec![
Span::styled(
format!("{}C", state.gpu.temperature_c),
Style::default().fg(temp_color),
),
Span::styled(" ", theme::muted_style()),
Span::styled(
format!("{}W / {}W", state.gpu.power_watts, state.gpu.power_limit_watts),
theme::muted_style(),
),
Span::styled(power_text, theme::muted_style()),
]))];
let info = List::new(info_items);
frame.render_widget(info, chunks[3]);

30
bin/fxt/src/tui/data_fetcher.rs
View File

@@ -43,6 +43,7 @@ pub enum StateUpdate {
sessions: Vec<TrainingSessionData>,
gpu: GpuData,
resources: SystemResources,
active_jobs: u32,
},
/// Service health statuses (from GetSystemStatus polling).
Services(Vec<ServiceData>),
@@ -170,7 +171,8 @@ impl DataFetcher {
ram_total_gb: f64::from(m.memory_total_mb) / 1024.0,
gpu_cluster_utilization: gpu.utilization,
};
let _ = tx.send(StateUpdate::TrainingMetrics { sessions, gpu, resources });
let active_jobs = m.active_k8s_jobs;
let _ = tx.send(StateUpdate::TrainingMetrics { sessions, gpu, resources, active_jobs });
}
Ok(None) => {
debug!("TrainingMetrics stream ended");
@@ -211,6 +213,8 @@ impl DataFetcher {
/// Service statuses — poll GetSystemStatus every 5s (the stream event
/// only carries SystemMetrics, not individual ServiceStatus entries).
/// After 3 consecutive failures, reports a connection error to indicate
/// stale data in the services cockpit.
fn spawn_system_status(&self, client: &FoxhuntClient) {
let mut mon = client.monitoring();
let tx = self.tx.clone();
@@ -219,6 +223,7 @@ impl DataFetcher {
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(5));
let mut consecutive_failures: u32 = 0;
loop {
tokio::select! {
_ = cancel.cancelled() => return,
@@ -230,6 +235,7 @@ impl DataFetcher {
});
match mon.get_system_status(req).await {
Ok(resp) => {
consecutive_failures = 0;
if !connected_sent.swap(true, Ordering::Relaxed) {
let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
}
@@ -242,7 +248,14 @@ impl DataFetcher {
let _ = tx.send(StateUpdate::Services(services));
}
Err(e) => {
debug!("SystemStatus poll failed: {e}");
consecutive_failures = consecutive_failures.saturating_add(1);
debug!("SystemStatus poll failed ({consecutive_failures}/3): {e}");
if consecutive_failures >= 3 {
warn!("SystemStatus: {consecutive_failures} consecutive failures, reporting stale");
let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Error(
format!("SystemStatus poll failed {consecutive_failures} times: {e}"),
)));
}
}
}
}
@@ -989,6 +1002,7 @@ fn convert_service_status(s: &monitoring::ServiceStatus) -> ServiceData {
uptime,
version: s.version.clone().unwrap_or_default(),
latency_ms,
error: s.error_message.clone(),
}
}
@@ -1191,8 +1205,8 @@ fn convert_download_job(j: &data_acquisition::DownloadJobSummary) -> DataFeedDat
DataFeedData {
symbol: j.symbols.join(","),
records_per_sec: 0, // not in summary
latency_us: 0,
records_per_sec: 0, // Not available in proto DownloadJobSummary
latency_us: 0, // Not available in proto DownloadJobSummary
status: status.to_owned(),
}
}
@@ -1223,12 +1237,12 @@ fn group_pods(pods: Vec<monitoring::PodInfo>) -> Vec<PodGroupData> {
groups
.into_iter()
.map(|(service, pods)| {
let ready_count = pods.iter().filter(|p| p.ready).count();
let total_count = pods.len();
.map(|(service, group_pods)| {
let ready_count = group_pods.iter().filter(|p| p.ready).count();
let total_count = group_pods.len();
PodGroupData {
service,
pods,
pods: group_pods,
ready_count,
total_count,
}

2
bin/fxt/src/tui/event_loop.rs
View File

@@ -253,10 +253,12 @@ fn apply_update(state: &mut AppState, update: StateUpdate) {
sessions,
gpu,
resources,
active_jobs,
} => {
state.training_sessions = sessions;
state.gpu = gpu;
state.system_resources = resources;
state.active_jobs = active_jobs;
}
StateUpdate::Services(services) => {
state.services = services;

5
bin/fxt/src/tui/state.rs
View File

@@ -37,6 +37,8 @@ pub struct AppState {
pub show_help: bool,
/// ML training sessions.
pub training_sessions: Vec<TrainingSessionData>,
/// Active K8s training jobs.
pub active_jobs: u32,
/// GPU telemetry.
pub gpu: GpuData,
/// Service health statuses.
@@ -81,6 +83,7 @@ impl Default for AppState {
current_cockpit: 0,
show_help: false,
training_sessions: Vec::new(),
active_jobs: 0,
gpu: GpuData::default(),
services: Vec::new(),
positions: Vec::new(),
@@ -155,6 +158,8 @@ pub struct ServiceData {
pub version: String,
/// Health-check round-trip latency in milliseconds.
pub latency_ms: f64,
/// Error reason when service is unhealthy.
pub error: Option<String>,
}
/// An open trading position.

174
bin/fxt/tests/error_tests.rs
View File

@@ -1,6 +1,6 @@
//! Unit tests for error module
//!
//! Tests all TliError variants and error handling functionality.
//! Tests all FxtError variants and error handling functionality.
// Suppress false-positive unused_crate_dependencies warnings
// dev-dependencies are shared across ALL test targets in the crate
@@ -8,127 +8,127 @@
#![allow(unused_crate_dependencies)]
use std::io;
use fxt::error::{TliError, TliResult};
use fxt::error::{FxtError, FxtResult};
/// Test TliError::Connection variant
/// Test FxtError::Connection variant
#[test]
fn test_tli_error_connection() {
let error = TliError::Connection("connection failed".to_string());
fn test_fxt_error_connection() {
let error = FxtError::Connection("connection failed".to_string());
assert!(matches!(error, TliError::Connection(_)));
assert!(matches!(error, FxtError::Connection(_)));
assert_eq!(error.to_string(), "Connection error: connection failed");
}
/// Test TliError::Config variant
/// Test FxtError::Config variant
#[test]
fn test_tli_error_config() {
let error = TliError::Config("invalid config".to_string());
fn test_fxt_error_config() {
let error = FxtError::Config("invalid config".to_string());
assert!(matches!(error, TliError::Config(_)));
assert!(matches!(error, FxtError::Config(_)));
assert_eq!(error.to_string(), "Configuration error: invalid config");
}
/// Test TliError::Dashboard variant
/// Test FxtError::Dashboard variant
#[test]
fn test_tli_error_dashboard() {
let error = TliError::Dashboard("render failed".to_string());
fn test_fxt_error_dashboard() {
let error = FxtError::Dashboard("render failed".to_string());
assert!(matches!(error, TliError::Dashboard(_)));
assert!(matches!(error, FxtError::Dashboard(_)));
assert_eq!(error.to_string(), "Dashboard error: render failed");
}
/// Test TliError::Io variant from std::io::Error
/// Test FxtError::Io variant from std::io::Error
#[test]
fn test_tli_error_io() {
fn test_fxt_error_io() {
let io_error = io::Error::new(io::ErrorKind::NotFound, "file not found");
let error = TliError::from(io_error);
let error = FxtError::from(io_error);
assert!(matches!(error, TliError::Io(_)));
assert!(matches!(error, FxtError::Io(_)));
assert!(error.to_string().contains("file not found"));
}
/// Test TliError::Grpc variant from tonic::Status
/// Test FxtError::Grpc variant from tonic::Status
#[test]
fn test_tli_error_grpc() {
fn test_fxt_error_grpc() {
let status = tonic::Status::unavailable("service unavailable");
let error = TliError::from(status);
let error = FxtError::from(status);
assert!(matches!(error, TliError::Grpc(_)));
assert!(matches!(error, FxtError::Grpc(_)));
assert!(error.to_string().contains("service unavailable"));
}
/// Test TliError::InvalidRequest variant
/// Test FxtError::InvalidRequest variant
#[test]
fn test_tli_error_invalid_request() {
let error = TliError::InvalidRequest("negative quantity".to_string());
fn test_fxt_error_invalid_request() {
let error = FxtError::InvalidRequest("negative quantity".to_string());
assert!(matches!(error, TliError::InvalidRequest(_)));
assert!(matches!(error, FxtError::InvalidRequest(_)));
assert_eq!(error.to_string(), "Invalid request: negative quantity");
}
/// Test TliError::NotFound variant
/// Test FxtError::NotFound variant
#[test]
fn test_tli_error_not_found() {
let error = TliError::NotFound("order not found".to_string());
fn test_fxt_error_not_found() {
let error = FxtError::NotFound("order not found".to_string());
assert!(matches!(error, TliError::NotFound(_)));
assert!(matches!(error, FxtError::NotFound(_)));
assert_eq!(error.to_string(), "Not found: order not found");
}
/// Test TliError::BufferFull variant
/// Test FxtError::BufferFull variant
#[test]
fn test_tli_error_buffer_full() {
let error = TliError::BufferFull("event buffer full".to_string());
fn test_fxt_error_buffer_full() {
let error = FxtError::BufferFull("event buffer full".to_string());
assert!(matches!(error, TliError::BufferFull(_)));
assert!(matches!(error, FxtError::BufferFull(_)));
assert_eq!(error.to_string(), "Buffer full: event buffer full");
}
/// Test TliError::InvalidSymbol variant
/// Test FxtError::InvalidSymbol variant
#[test]
fn test_tli_error_invalid_symbol() {
let error = TliError::InvalidSymbol("INVALID@123".to_string());
fn test_fxt_error_invalid_symbol() {
let error = FxtError::InvalidSymbol("INVALID@123".to_string());
assert!(matches!(error, TliError::InvalidSymbol(_)));
assert!(matches!(error, FxtError::InvalidSymbol(_)));
assert_eq!(error.to_string(), "Invalid symbol: INVALID@123");
}
/// Test TliError::Other variant
/// Test FxtError::Other variant
#[test]
fn test_tli_error_other() {
let error = TliError::Other("unexpected error".to_string());
fn test_fxt_error_other() {
let error = FxtError::Other("unexpected error".to_string());
assert!(matches!(error, TliError::Other(_)));
assert!(matches!(error, FxtError::Other(_)));
assert_eq!(error.to_string(), "Other error: unexpected error");
}
/// Test TliError debug formatting
/// Test FxtError debug formatting
#[test]
fn test_tli_error_debug() {
let error = TliError::Connection("test".to_string());
fn test_fxt_error_debug() {
let error = FxtError::Connection("test".to_string());
let debug_str = format!("{:?}", error);
assert!(debug_str.contains("Connection"));
assert!(debug_str.contains("test"));
}
/// Test TliResult with Ok value
/// Test FxtResult with Ok value
#[test]
fn test_tli_result_ok() {
let result: TliResult<i32> = Ok(42);
fn test_fxt_result_ok() {
let result: FxtResult<i32> = Ok(42);
assert!(result.is_ok());
assert_eq!(result.unwrap(), 42);
}
/// Test TliResult with Err value
/// Test FxtResult with Err value
#[test]
fn test_tli_result_err() {
let result: TliResult<i32> = Err(TliError::Connection("failed".to_string()));
fn test_fxt_result_err() {
let result: FxtResult<i32> = Err(FxtError::Connection("failed".to_string()));
assert!(result.is_err());
let error = result.unwrap_err();
assert!(matches!(error, TliError::Connection(_)));
assert!(matches!(error, FxtError::Connection(_)));
}
/// Test error message formatting
@@ -136,19 +136,19 @@ fn test_tli_result_err() {
fn test_error_message_formatting() {
let errors = vec![
(
TliError::Connection("timeout".to_string()),
FxtError::Connection("timeout".to_string()),
"Connection error: timeout",
),
(
TliError::Config("missing field".to_string()),
FxtError::Config("missing field".to_string()),
"Configuration error: missing field",
),
(
TliError::InvalidRequest("invalid data".to_string()),
FxtError::InvalidRequest("invalid data".to_string()),
"Invalid request: invalid data",
),
(
TliError::NotFound("resource".to_string()),
FxtError::NotFound("resource".to_string()),
"Not found: resource",
),
];
@@ -161,7 +161,7 @@ fn test_error_message_formatting() {
/// Test error with empty message
#[test]
fn test_error_with_empty_message() {
let error = TliError::Connection("".to_string());
let error = FxtError::Connection("".to_string());
assert_eq!(error.to_string(), "Connection error: ");
}
@@ -169,7 +169,7 @@ fn test_error_with_empty_message() {
#[test]
fn test_error_with_long_message() {
let long_message = "a".repeat(10000);
let error = TliError::Other(long_message.clone());
let error = FxtError::Other(long_message.clone());
assert!(error.to_string().contains(&long_message));
}
@@ -178,7 +178,7 @@ fn test_error_with_long_message() {
#[test]
fn test_error_with_special_characters() {
let message = "Error: timeout (500ms) at server://host:8080";
let error = TliError::Connection(message.to_string());
let error = FxtError::Connection(message.to_string());
assert!(error.to_string().contains(message));
}
@@ -195,9 +195,9 @@ fn test_error_from_different_io_kinds() {
for kind in kinds {
let io_error = io::Error::new(kind, "test error");
let tli_error = TliError::from(io_error);
let fxt_error = FxtError::from(io_error);
assert!(matches!(tli_error, TliError::Io(_)));
assert!(matches!(fxt_error, FxtError::Io(_)));
}
}
@@ -213,19 +213,19 @@ fn test_error_from_different_grpc_codes() {
];
for status in statuses {
let tli_error = TliError::from(status);
assert!(matches!(tli_error, TliError::Grpc(_)));
let fxt_error = FxtError::from(status);
assert!(matches!(fxt_error, FxtError::Grpc(_)));
}
}
/// Test error propagation in functions
#[test]
fn test_error_propagation() {
fn operation_that_fails() -> TliResult<()> {
Err(TliError::Connection("failed".to_string()))
fn operation_that_fails() -> FxtResult<()> {
Err(FxtError::Connection("failed".to_string()))
}
fn caller() -> TliResult<()> {
fn caller() -> FxtResult<()> {
operation_that_fails()?;
Ok(())
}
@@ -238,18 +238,18 @@ fn test_error_propagation() {
#[test]
fn test_error_pattern_matching() {
let errors = vec![
TliError::Connection("test".to_string()),
TliError::Config("test".to_string()),
TliError::InvalidRequest("test".to_string()),
TliError::NotFound("test".to_string()),
FxtError::Connection("test".to_string()),
FxtError::Config("test".to_string()),
FxtError::InvalidRequest("test".to_string()),
FxtError::NotFound("test".to_string()),
];
for error in errors {
match error {
TliError::Connection(_) => assert!(true),
TliError::Config(_) => assert!(true),
TliError::InvalidRequest(_) => assert!(true),
TliError::NotFound(_) => assert!(true),
FxtError::Connection(_) => assert!(true),
FxtError::Config(_) => assert!(true),
FxtError::InvalidRequest(_) => assert!(true),
FxtError::NotFound(_) => assert!(true),
_ => assert!(false, "unexpected variant"),
}
}
@@ -259,7 +259,7 @@ fn test_error_pattern_matching() {
#[test]
fn test_error_with_newlines() {
let message = "Line 1\nLine 2\nLine 3";
let error = TliError::Other(message.to_string());
let error = FxtError::Other(message.to_string());
assert!(error.to_string().contains("Line 1"));
assert!(error.to_string().contains("Line 2"));
@@ -269,7 +269,7 @@ fn test_error_with_newlines() {
#[test]
fn test_error_with_unicode() {
let message = "Error: 失败 (Chinese), エラー (Japanese), ошибка (Russian)";
let error = TliError::Connection(message.to_string());
let error = FxtError::Connection(message.to_string());
assert!(error.to_string().contains("失败"));
assert!(error.to_string().contains("エラー"));
@@ -278,11 +278,11 @@ fn test_error_with_unicode() {
/// Test Result type aliases
#[test]
fn test_result_type_alias() {
fn returns_tli_result() -> TliResult<String> {
fn returns_fxt_result() -> FxtResult<String> {
Ok("success".to_string())
}
let result = returns_tli_result();
let result = returns_fxt_result();
assert!(result.is_ok());
assert_eq!(result.unwrap(), "success");
}
@@ -294,14 +294,14 @@ fn test_error_chain() {
Err(io::Error::new(io::ErrorKind::NotFound, "inner error"))
}
fn outer_operation() -> TliResult<()> {
inner_operation().map_err(|e| TliError::from(e))?;
fn outer_operation() -> FxtResult<()> {
inner_operation().map_err(|e| FxtError::from(e))?;
Ok(())
}
let result = outer_operation();
assert!(result.is_err());
assert!(matches!(result.unwrap_err(), TliError::Io(_)));
assert!(matches!(result.unwrap_err(), FxtError::Io(_)));
}
/// Test InvalidSymbol with various invalid symbols
@@ -317,8 +317,8 @@ fn test_invalid_symbol_variants() {
];
for symbol in invalid_symbols {
let error = TliError::InvalidSymbol(symbol.to_string());
assert!(matches!(error, TliError::InvalidSymbol(_)));
let error = FxtError::InvalidSymbol(symbol.to_string());
assert!(matches!(error, FxtError::InvalidSymbol(_)));
}
}
@@ -333,15 +333,15 @@ fn test_buffer_full_variants() {
];
for buffer_type in buffer_types {
let error = TliError::BufferFull(format!("{} full", buffer_type));
assert!(matches!(error, TliError::BufferFull(_)));
let error = FxtError::BufferFull(format!("{} full", buffer_type));
assert!(matches!(error, FxtError::BufferFull(_)));
}
}
/// Test error unwrap_or_else with fallback
#[test]
fn test_error_unwrap_or_else() {
let result: TliResult<i32> = Err(TliError::NotFound("test".to_string()));
let result: FxtResult<i32> = Err(FxtError::NotFound("test".to_string()));
let value = result.unwrap_or_else(|_| 42);
assert_eq!(value, 42);
@@ -350,7 +350,7 @@ fn test_error_unwrap_or_else() {
/// Test error map_err transformation
#[test]
fn test_error_map_err() {
let result: TliResult<i32> = Err(TliError::Connection("test".to_string()));
let result: FxtResult<i32> = Err(FxtError::Connection("test".to_string()));
let transformed = result.map_err(|e| format!("Transformed: {}", e));
assert!(transformed.is_err());

1570
bin/fxt/tests/integration/end_to_end_tests.rs
View File

File diff suppressed because it is too large Load Diff

1302
bin/fxt/tests/integration/error_handling_tests.rs
View File

File diff suppressed because it is too large Load Diff

2
bin/fxt/tests/integration/mod.rs
View File

@@ -1,3 +1,3 @@
//! Integration tests module - all integration tests disabled
//!
//! Tests need refactoring after TLI client architecture changes
//! Tests need refactoring after FXT client architecture changes

6
bin/fxt/tests/integration/performance_tests.rs
View File

@@ -144,7 +144,7 @@ impl PerformanceTestEnvironment {
}
/// Setup high-performance test environment
pub async fn setup(&mut self) -> TliResult<()> {
pub async fn setup(&mut self) -> FxtResult<()> {
tracing::info!("Setting up performance test environment");
// Start high-performance mock servers
@@ -185,7 +185,7 @@ impl PerformanceTestEnvironment {
}
/// Cleanup test environment
pub async fn teardown(&mut self) -> TliResult<()> {
pub async fn teardown(&mut self) -> FxtResult<()> {
tracing::info!("Tearing down performance test environment");
// Shutdown clients
@@ -936,7 +936,7 @@ mod stress_tests {
let success = match &result {
Ok(_) => true,
Err(TliError::Connection(_)) => {
Err(FxtError::Connection(_)) => {
error_counter.fetch_add(1, Ordering::Relaxed);
false
}

8
bin/fxt/tests/integration/service_integration_tests.rs
View File

@@ -31,8 +31,8 @@ pub struct ServiceIntegrationTests {
/// Mock server trait for test servers
pub trait MockServer: Send + Sync {
fn start(&mut self) -> TliResult<u16>;
fn stop(&mut self) -> TliResult<()>;
fn start(&mut self) -> FxtResult<u16>;
fn stop(&mut self) -> FxtResult<()>;
fn is_running(&self) -> bool;
fn get_port(&self) -> Option<u16>;
}
@@ -48,7 +48,7 @@ impl ServiceIntegrationTests {
}
/// Setup test environment with mock services
pub async fn setup(&mut self) -> TliResult<()> {
pub async fn setup(&mut self) -> FxtResult<()> {
tracing::info!("Setting up service integration test environment");
// Start mock trading service
@@ -88,7 +88,7 @@ impl ServiceIntegrationTests {
}
/// Cleanup test environment
pub async fn teardown(&mut self) -> TliResult<()> {
pub async fn teardown(&mut self) -> FxtResult<()> {
tracing::info!("Tearing down service integration test environment");
// Shutdown clients

2
bin/fxt/tests/ml_trading_commands_test.rs
View File

@@ -142,7 +142,7 @@ mod test_auth {
let encryption_key = hex::encode([42u8; 32]); // Simple deterministic key for tests
std::env::set_var("FOXHUNT_ENCRYPTION_KEY", &encryption_key);
// Create temp directory structure: temp/foxhunt-tli/tokens/
// Create temp directory structure: temp/foxhunt-fxt/tokens/
let temp_base =
std::env::temp_dir().join(format!("foxhunt_fxt_config_{}", std::process::id()));
let config_home = temp_base.clone();

81
proto/config_service.proto
View File

@@ -1,81 +0,0 @@
syntax = "proto3";
package foxhunt.config;
// Configuration Management Service
service ConfigurationService {
// Get a single configuration value
rpc GetConfig(GetConfigRequest) returns (GetConfigResponse);
// Update a configuration value
rpc UpdateConfig(UpdateConfigRequest) returns (UpdateConfigResponse);
// List all configurations for a service scope
rpc ListConfigs(ListConfigsRequest) returns (ListConfigsResponse);
// Trigger configuration reload
rpc ReloadConfig(ReloadConfigRequest) returns (ReloadConfigResponse);
}
// Get configuration request
message GetConfigRequest {
string service_scope = 1;
string config_key = 2;
}
// Get configuration response
message GetConfigResponse {
string config_value = 1; // JSON-serialized value
string data_type = 2;
string description = 3;
int64 updated_at = 4; // Unix timestamp
string updated_by = 5;
}
// Update configuration request
message UpdateConfigRequest {
string service_scope = 1;
string config_key = 2;
string new_value = 3; // JSON-serialized value
string updated_by = 4;
}
// Update configuration response
message UpdateConfigResponse {
bool success = 1;
string message = 2;
}
// List configurations request
message ListConfigsRequest {
optional string service_scope = 1; // If not provided, lists all scopes
}
// Configuration item
message ConfigItem {
string service_scope = 1;
string config_key = 2;
string config_value = 3; // JSON-serialized value
string data_type = 4;
string description = 5;
int64 created_at = 6;
int64 updated_at = 7;
string updated_by = 8;
}
// List configurations response
message ListConfigsResponse {
repeated ConfigItem configs = 1;
}
// Reload configuration request
message ReloadConfigRequest {
optional string service_scope = 1;
optional string config_key = 2;
}
// Reload configuration response
message ReloadConfigResponse {
bool success = 1;
string message = 2;
}

17
services/api_gateway/build.rs
View File

@@ -4,10 +4,9 @@
//! Proto packages:
//! - `foxhunt.tli` (fxt_trading.proto) - Fat-client unified interface (server+client)
//! - `trading` (trading.proto) - Backend trading service (client-only)
//! - `foxhunt.config` (config_service.proto) - Config service (server+client)
//! - `risk` (risk.proto) - Risk service (client-only)
//! - `monitoring` (monitoring.proto) - Monitoring service (server+client)
//! - `config` (config.proto) - Config backend (client-only)
//! - `config` (config.proto) - Config service (server+client)
//! - `ml_training` (ml_training.proto) - ML Training service (server+client)
//! - `trading_agent` (trading_agent.proto) - Trading Agent service (server+client)
@@ -17,19 +16,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let out_dir = std::path::PathBuf::from(std::env::var("OUT_DIR")?);
// Compile Config Service proto (foxhunt.config package)
config
.clone()
.build_server(true)
.build_client(true)
.file_descriptor_set_path(out_dir.join("config_service_descriptor.bin"))
.compile_well_known_types(true)
.extern_path(".google.protobuf", "::prost_types")
.type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]")
.server_mod_attribute(".", "#[allow(unused_qualifications)]")
.client_mod_attribute(".", "#[allow(unused_qualifications)]")
.compile_protos(&["../../proto/config_service.proto"], &["../../proto"])?;
// Compile fat-client TLI proto (package: foxhunt.tli) — server+client
// Contains TradingService, BacktestingService, and MLService
// API Gateway acts as server for incoming client requests
@@ -193,7 +179,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
&["../../proto"],
)?;
println!("cargo:rerun-if-changed=../../proto/config_service.proto");
println!("cargo:rerun-if-changed=../../proto/fxt_trading.proto");
println!("cargo:rerun-if-changed=../../proto/trading.proto");
println!("cargo:rerun-if-changed=../../proto/risk.proto");

304
services/api_gateway/src/config/endpoints.rs
View File

@@ -2,24 +2,35 @@
use crate::config::ConfigurationManager;
use crate::error::GatewayConfigError;
use futures::Stream;
use std::pin::Pin;
use tonic::{Request, Response, Status};
use tracing::{debug, info, instrument};
// Import generated proto code from crate root
use crate::foxhunt::config_proto::{
configuration_service_server::{ConfigurationService, ConfigurationServiceServer},
ConfigItem as ProtoConfigItem, GetConfigRequest, GetConfigResponse, ListConfigsRequest,
ListConfigsResponse, ReloadConfigRequest, ReloadConfigResponse, UpdateConfigRequest,
UpdateConfigResponse,
use crate::config_backend::{
config_service_server::{ConfigService, ConfigServiceServer},
ConfigChangeEvent, ConfigDataType, ConfigurationCategory, ConfigurationSetting,
DeleteConfigurationRequest, DeleteConfigurationResponse, ExportConfigurationRequest,
ExportConfigurationResponse, GetConfigSchemaRequest, GetConfigSchemaResponse,
GetConfigurationHistoryRequest, GetConfigurationHistoryResponse, GetConfigurationRequest,
GetConfigurationResponse, ImportConfigurationRequest, ImportConfigurationResponse,
ListCategoriesRequest, ListCategoriesResponse, RollbackConfigurationRequest,
RollbackConfigurationResponse, StreamConfigChangesRequest, UpdateConfigSchemaRequest,
UpdateConfigSchemaResponse, UpdateConfigurationRequest, UpdateConfigurationResponse,
ValidateConfigurationRequest, ValidateConfigurationResponse,
};
/// gRPC service implementation for configuration management
pub struct ConfigurationServiceImpl {
/// gRPC service implementation for gateway-local configuration management.
///
/// Implements the `config.ConfigService` trait from config.proto.
/// Handles GetConfiguration, UpdateConfiguration, and ListCategories locally;
/// all other RPCs return `Status::unimplemented`.
pub struct GatewayConfigService {
manager: std::sync::Arc<tokio::sync::RwLock<ConfigurationManager>>,
}
impl ConfigurationServiceImpl {
/// Creates a new ConfigurationServiceImpl
impl GatewayConfigService {
/// Creates a new GatewayConfigService
pub fn new(manager: ConfigurationManager) -> Self {
Self {
manager: std::sync::Arc::new(tokio::sync::RwLock::new(manager)),
@@ -27,63 +38,94 @@ impl ConfigurationServiceImpl {
}
/// Returns the gRPC server instance
pub fn into_server(self) -> ConfigurationServiceServer<Self> {
ConfigurationServiceServer::new(self)
pub fn into_server(self) -> ConfigServiceServer<Self> {
ConfigServiceServer::new(self)
}
}
#[tonic::async_trait]
impl ConfigurationService for ConfigurationServiceImpl {
impl ConfigService for GatewayConfigService {
type StreamConfigChangesStream =
Pin<Box<dyn Stream<Item = Result<ConfigChangeEvent, Status>> + Send>>;
#[instrument(skip_all)]
async fn get_config(
async fn get_configuration(
&self,
request: Request<GetConfigRequest>,
) -> Result<Response<GetConfigResponse>, Status> {
request: Request<GetConfigurationRequest>,
) -> Result<Response<GetConfigurationResponse>, Status> {
let req = request.into_inner();
debug!("GetConfig: {}/{}", req.service_scope, req.config_key);
let category = req.category.unwrap_or_default();
let key = req.key.unwrap_or_default();
debug!("GetConfiguration: {}/{}", category, key);
if category.is_empty() && key.is_empty() {
// List all configs
let manager = self.manager.read().await;
let configs = manager
.list_configs(None)
.await
.map_err(|e| Status::internal(format!("{}", e)))?;
let settings: Result<Vec<ConfigurationSetting>, Status> = configs
.into_iter()
.map(|c| config_to_setting(&c))
.collect();
return Ok(Response::new(GetConfigurationResponse {
settings: settings?,
}));
}
let manager = self.manager.read().await;
let config = manager
.get_config(&req.service_scope, &req.config_key)
.await
.map_err(|e| match e {
GatewayConfigError::NotFound { .. } => Status::not_found(format!("{}", e)),
_ => Status::internal(format!("{}", e)),
})?;
Ok(Response::new(GetConfigResponse {
config_value: serde_json::to_string(&config.config_value)
.map_err(|e| Status::internal(format!("Serialization error: {}", e)))?,
data_type: config.data_type,
description: config.description.unwrap_or_default(),
updated_at: config.updated_at.timestamp(),
updated_by: config.updated_by.unwrap_or_default(),
}))
if !key.is_empty() {
// Get specific config
let config = manager
.get_config(&category, &key)
.await
.map_err(|e| match e {
GatewayConfigError::NotFound { .. } => Status::not_found(format!("{}", e)),
_ => Status::internal(format!("{}", e)),
})?;
Ok(Response::new(GetConfigurationResponse {
settings: vec![config_to_setting(&config)?],
}))
} else {
// List by category
let configs = manager
.list_configs(Some(&category))
.await
.map_err(|e| Status::internal(format!("{}", e)))?;
let settings: Result<Vec<ConfigurationSetting>, Status> = configs
.into_iter()
.map(|c| config_to_setting(&c))
.collect();
Ok(Response::new(GetConfigurationResponse {
settings: settings?,
}))
}
}
#[instrument(skip_all)]
async fn update_config(
async fn update_configuration(
&self,
request: Request<UpdateConfigRequest>,
) -> Result<Response<UpdateConfigResponse>, Status> {
request: Request<UpdateConfigurationRequest>,
) -> Result<Response<UpdateConfigurationResponse>, Status> {
let req = request.into_inner();
info!(
"UpdateConfig: {}/{} by {}",
req.service_scope, req.config_key, req.updated_by
"UpdateConfiguration: {}/{} by {}",
req.category, req.key, req.changed_by
);
// Parse new value as JSON
let new_value: serde_json::Value = serde_json::from_str(&req.new_value)
let new_value: serde_json::Value = serde_json::from_str(&req.value)
.map_err(|e| Status::invalid_argument(format!("Invalid JSON: {}", e)))?;
let manager = self.manager.read().await;
manager
.update_config(
&req.service_scope,
&req.config_key,
new_value,
&req.updated_by,
)
.update_config(&req.category, &req.key, new_value, &req.changed_by)
.await
.map_err(|e| match e {
GatewayConfigError::Validation(msg) => Status::invalid_argument(msg),
@@ -91,65 +133,159 @@ impl ConfigurationService for ConfigurationServiceImpl {
_ => Status::internal(format!("{}", e)),
})?;
Ok(Response::new(UpdateConfigResponse {
Ok(Response::new(UpdateConfigurationResponse {
success: true,
message: format!(
"Successfully updated {}/{}",
req.service_scope, req.config_key
),
message: format!("Successfully updated {}/{}", req.category, req.key),
validation_result: None,
timestamp: chrono::Utc::now().timestamp(),
}))
}
#[instrument(skip_all)]
async fn list_configs(
async fn delete_configuration(
&self,
request: Request<ListConfigsRequest>,
) -> Result<Response<ListConfigsResponse>, Status> {
_request: Request<DeleteConfigurationRequest>,
) -> Result<Response<DeleteConfigurationResponse>, Status> {
Err(Status::unimplemented(
"Delete not supported by gateway config",
))
}
#[instrument(skip_all)]
async fn list_categories(
&self,
request: Request<ListCategoriesRequest>,
) -> Result<Response<ListCategoriesResponse>, Status> {
let req = request.into_inner();
debug!("ListConfigs: {:?}", req.service_scope);
debug!("ListCategories: {:?}", req.parent_category);
let manager = self.manager.read().await;
let configs = manager
.list_configs(req.service_scope.as_deref())
.list_configs(req.parent_category.as_deref())
.await
.map_err(|e| Status::internal(format!("{}", e)))?;
let proto_configs: Result<Vec<ProtoConfigItem>, Status> = configs
.into_iter()
.map(|c| {
Ok(ProtoConfigItem {
service_scope: c.service_scope,
config_key: c.config_key,
config_value: serde_json::to_string(&c.config_value)
.map_err(|e| Status::internal(format!("{}", e)))?,
data_type: c.data_type,
description: c.description.unwrap_or_default(),
created_at: c.created_at.timestamp(),
updated_at: c.updated_at.timestamp(),
updated_by: c.updated_by.unwrap_or_default(),
})
// Extract unique categories from configs
let mut seen = std::collections::HashSet::new();
let categories = configs
.iter()
.filter_map(|c| {
if seen.insert(c.service_scope.clone()) {
Some(ConfigurationCategory {
id: 0,
name: c.service_scope.clone(),
description: String::new(),
parent_id: None,
display_order: 0,
icon: None,
created_at: c.created_at.timestamp(),
children: vec![],
})
} else {
None
}
})
.collect();
Ok(Response::new(ListConfigsResponse {
configs: proto_configs?,
}))
Ok(Response::new(ListCategoriesResponse { categories }))
}
#[instrument(skip_all)]
async fn reload_config(
async fn stream_config_changes(
&self,
request: Request<ReloadConfigRequest>,
) -> Result<Response<ReloadConfigResponse>, Status> {
let req = request.into_inner();
info!("ReloadConfig: {:?}/{:?}", req.service_scope, req.config_key);
_request: Request<StreamConfigChangesRequest>,
) -> Result<Response<Self::StreamConfigChangesStream>, Status> {
Err(Status::unimplemented(
"Streaming not supported by gateway config",
))
}
// For now, this is a no-op since hot-reload is automatic via NOTIFY/LISTEN
// In the future, this could force a cache invalidation
async fn validate_configuration(
&self,
_request: Request<ValidateConfigurationRequest>,
) -> Result<Response<ValidateConfigurationResponse>, Status> {
Err(Status::unimplemented(
"Validation not supported by gateway config",
))
}
Ok(Response::new(ReloadConfigResponse {
success: true,
message: "Configuration reload triggered (hot-reload active)".to_string(),
}))
async fn get_configuration_history(
&self,
_request: Request<GetConfigurationHistoryRequest>,
) -> Result<Response<GetConfigurationHistoryResponse>, Status> {
Err(Status::unimplemented(
"History not supported by gateway config",
))
}
async fn rollback_configuration(
&self,
_request: Request<RollbackConfigurationRequest>,
) -> Result<Response<RollbackConfigurationResponse>, Status> {
Err(Status::unimplemented(
"Rollback not supported by gateway config",
))
}
async fn export_configuration(
&self,
_request: Request<ExportConfigurationRequest>,
) -> Result<Response<ExportConfigurationResponse>, Status> {
Err(Status::unimplemented(
"Export not supported by gateway config",
))
}
async fn import_configuration(
&self,
_request: Request<ImportConfigurationRequest>,
) -> Result<Response<ImportConfigurationResponse>, Status> {
Err(Status::unimplemented(
"Import not supported by gateway config",
))
}
async fn get_config_schema(
&self,
_request: Request<GetConfigSchemaRequest>,
) -> Result<Response<GetConfigSchemaResponse>, Status> {
Err(Status::unimplemented(
"Schema not supported by gateway config",
))
}
async fn update_config_schema(
&self,
_request: Request<UpdateConfigSchemaRequest>,
) -> Result<Response<UpdateConfigSchemaResponse>, Status> {
Err(Status::unimplemented(
"Schema update not supported by gateway config",
))
}
}
/// Convert a ConfigItem (from ConfigurationManager) to a ConfigurationSetting proto message
fn config_to_setting(c: &crate::config::ConfigItem) -> Result<ConfigurationSetting, Status> {
Ok(ConfigurationSetting {
id: 0,
category: c.service_scope.clone(),
key: c.config_key.clone(),
value: serde_json::to_string(&c.config_value)
.map_err(|e| Status::internal(format!("Serialization error: {}", e)))?,
data_type: ConfigDataType::String as i32,
hot_reload: false,
description: c.description.clone().unwrap_or_default(),
default_value: None,
required: false,
sensitive: false,
validation_rule: None,
environment_override: None,
min_value: None,
max_value: None,
enum_values: None,
depends_on: vec![],
tags: vec![],
display_order: 0,
created_at: c.created_at.timestamp(),
modified_at: c.updated_at.timestamp(),
})
}

2
services/api_gateway/src/config/mod.rs
View File

@@ -5,7 +5,7 @@ pub mod endpoints;
pub mod manager;
pub mod validator;
pub use endpoints::ConfigurationServiceImpl;
pub use endpoints::GatewayConfigService;
pub use manager::{ConfigItem, ConfigurationManager};
pub use validator::{ConfigValidator, ValidationRules};

40
services/api_gateway/src/grpc/monitoring_handler.rs
View File

@@ -21,7 +21,7 @@ use tracing::{error, info, instrument};
use crate::monitoring::monitoring_service_server::{MonitoringService, MonitoringServiceServer};
use crate::monitoring::{
AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent, AlertEventType,
AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent,
ClusterPodsResponse, EpochFinancialSnapshot, GetActiveAlertsRequest, GetActiveAlertsResponse,
GetEpochHistoryRequest, GetEpochHistoryResponse, GetHealthCheckRequest,
GetHealthCheckResponse, GetLatencyMetricsRequest, GetLatencyMetricsResponse,
@@ -625,6 +625,8 @@ impl MonitoringService for MonitoringServiceHandler {
}
);
let prom = self.prom.clone();
let stream = async_stream::stream! {
let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
loop {
@@ -644,7 +646,10 @@ impl MonitoringService for MonitoringServiceHandler {
})
.collect();
let statuses = futures::future::join_all(checks).await;
let (statuses, (cpu, mem, disk)) = tokio::join!(
futures::future::join_all(checks),
prom.query_system_metrics(),
);
let healthy_count = statuses
.iter()
@@ -682,7 +687,12 @@ impl MonitoringService for MonitoringServiceHandler {
total_services: total as i32,
critical_issues,
system_uptime_seconds: start_time.elapsed().as_secs() as i64,
system_metrics: Some(SystemMetrics::default()),
system_metrics: Some(SystemMetrics {
cpu_usage_percent: cpu,
memory_usage_percent: mem,
disk_usage_percent: disk,
..Default::default()
}),
};
let event = SystemStatusEvent {
@@ -903,27 +913,9 @@ impl MonitoringService for MonitoringServiceHandler {
request: Request<crate::monitoring::StreamAlertsRequest>,
) -> Result<Response<Self::StreamAlertsStream>, Status> {
let _req = request.into_inner();
info!("StreamAlerts: interval=10s (placeholder, no alertmanager integration yet)");
let stream = async_stream::stream! {
let mut interval = tokio::time::interval(Duration::from_secs(10));
loop {
interval.tick().await;
// Placeholder: yields an empty AlertEvent each tick.
// Will be wired to Prometheus Alertmanager when available.
let event = AlertEvent {
alert: None,
event_type: AlertEventType::Unspecified.into(),
timestamp: chrono::Utc::now().timestamp(),
};
yield Ok(event);
}
};
Ok(Response::new(Box::pin(stream)))
Err(Status::unimplemented(
"StreamAlerts will be wired to Prometheus Alertmanager",
))
}
async fn acknowledge_alert(

7
services/api_gateway/src/lib.rs
View File

@@ -26,8 +26,6 @@ pub const ML_TRAINING_FILE_DESCRIPTOR_SET: &[u8] =
tonic::include_file_descriptor_set!("ml_training_descriptor");
pub const TRADING_AGENT_FILE_DESCRIPTOR_SET: &[u8] =
tonic::include_file_descriptor_set!("trading_agent_descriptor");
pub const CONFIG_SERVICE_FILE_DESCRIPTOR_SET: &[u8] =
tonic::include_file_descriptor_set!("config_service_descriptor");
pub const TRADING_FILE_DESCRIPTOR_SET: &[u8] =
tonic::include_file_descriptor_set!("trading_descriptor");
pub const RISK_FILE_DESCRIPTOR_SET: &[u8] =
@@ -46,9 +44,6 @@ pub mod foxhunt {
pub mod tli {
tonic::include_proto!("foxhunt.tli");
}
pub mod config_proto {
tonic::include_proto!("foxhunt.config");
}
}
pub mod ml_training {
@@ -113,7 +108,7 @@ pub use error::{GatewayConfigError, GatewayConfigResult};
// Re-export configuration management types
pub use config::{
AuthzMetrics, AuthzService, ConfigItem, ConfigValidator, ConfigurationManager,
ConfigurationServiceImpl, PermissionResult,
GatewayConfigService, PermissionResult,
};
// Re-export routing and rate limiting types

1
services/api_gateway/src/main.rs
View File

@@ -682,7 +682,6 @@ async fn main() -> Result<()> {
.register_encoded_file_descriptor_set(api_gateway::MONITORING_FILE_DESCRIPTOR_SET)
.register_encoded_file_descriptor_set(api_gateway::ML_TRAINING_FILE_DESCRIPTOR_SET)
.register_encoded_file_descriptor_set(api_gateway::TRADING_AGENT_FILE_DESCRIPTOR_SET)
.register_encoded_file_descriptor_set(api_gateway::CONFIG_SERVICE_FILE_DESCRIPTOR_SET)
.register_encoded_file_descriptor_set(api_gateway::TRADING_FILE_DESCRIPTOR_SET)
.register_encoded_file_descriptor_set(api_gateway::RISK_FILE_DESCRIPTOR_SET)
.register_encoded_file_descriptor_set(api_gateway::CONFIG_FILE_DESCRIPTOR_SET)

4
testing/integration/fixtures/mod.rs
View File

@@ -39,7 +39,7 @@ use tokio::sync::RwLock;
use uuid::Uuid;
// Import TLI types explicitly (tli crate is available as dependency)
use fxt::error::TliResult;
use fxt::error::FxtResult;
// Re-export sub-modules for easy access
pub mod builders;
@@ -788,7 +788,7 @@ impl std::fmt::Debug for TestEnvironment {
}
impl TestEnvironment {
pub async fn new(config: IntegrationTestConfig) -> TliResult<Self> {
pub async fn new(config: IntegrationTestConfig) -> FxtResult<Self> {
let metrics = Arc::new(TestMetricsCollector::new());
let port_manager = Arc::new(TestPortManager::new());

36
testing/integration/mocks/mod.rs
View File

@@ -189,8 +189,8 @@ impl MockTradingService {
/// # Returns
/// * `Ok(MockTradingService)` - Configured mock service ready to start
///
/// * `Err(TliError)` - If port allocation or initialization failed
pub async fn new() -> TliResult<Self> {
/// * `Err(FxtError)` - If port allocation or initialization failed
pub async fn new() -> FxtResult<Self> {
Self::new_with_config(MockServiceConfig::default()).await
}
@@ -202,8 +202,8 @@ impl MockTradingService {
/// # Returns
/// * `Ok(MockTradingService)` - Configured mock service
///
/// * `Err(TliError)` - If port allocation failed
pub async fn new_with_config(config: MockServiceConfig) -> TliResult<Self> {
/// * `Err(FxtError)` - If port allocation failed
pub async fn new_with_config(config: MockServiceConfig) -> FxtResult<Self> {
let port = TEST_PORT_MANAGER.allocate_port().await;
Ok(Self {
@@ -266,8 +266,8 @@ impl MockTradingService {
/// # Returns
/// * `Ok(())` - Service started successfully
///
/// * `Err(TliError)` - If service failed to start
pub async fn start(&mut self) -> TliResult<()> {
/// * `Err(FxtError)` - If service failed to start
pub async fn start(&mut self) -> FxtResult<()> {
let port = self.port;
let config = self.config.clone();
let order_responses = Arc::clone(&self.order_responses);
@@ -412,7 +412,7 @@ pub struct CreateBacktestResponse {
impl MockBacktestingService {
/// Create new mock backtesting service
pub async fn new() -> TliResult<Self> {
pub async fn new() -> FxtResult<Self> {
let port = TEST_PORT_MANAGER.allocate_port().await;
Ok(Self {
@@ -498,15 +498,15 @@ impl Default for TestDatabaseConfig {
impl TestDatabaseManager {
/// Create new test database manager
pub async fn new() -> TliResult<Self> {
pub async fn new() -> FxtResult<Self> {
Self::new_with_config(TestDatabaseConfig::default()).await
}
/// Create new test database manager with custom configuration
pub async fn new_with_config(config: TestDatabaseConfig) -> TliResult<Self> {
pub async fn new_with_config(config: TestDatabaseConfig) -> FxtResult<Self> {
let pool = sqlx::PgPool::connect(&config.database_url)
.await
.map_err(|e| TliError::DatabaseError(format!("Failed to connect to test database: {}", e)))?;
.map_err(|e| FxtError::DatabaseError(format!("Failed to connect to test database: {}", e)))?;
Ok(Self { pool, config })
}
@@ -517,18 +517,18 @@ impl TestDatabaseManager {
}
/// Verify order was persisted
pub async fn verify_order_persisted(&self, order_id: &str) -> TliResult<bool> {
pub async fn verify_order_persisted(&self, order_id: &str) -> FxtResult<bool> {
let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM orders WHERE order_id = $1")
.bind(order_id)
.fetch_one(&self.pool)
.await
.map_err(|e| TliError::DatabaseError(format!("Failed to verify order persistence: {}", e)))?;
.map_err(|e| FxtError::DatabaseError(format!("Failed to verify order persistence: {}", e)))?;
Ok(count > 0)
}
/// Clean up test data
pub async fn cleanup(&self) -> TliResult<()> {
pub async fn cleanup(&self) -> FxtResult<()> {
if self.config.enable_cleanup {
let cleanup_queries = vec![
"DELETE FROM trading_events WHERE source LIKE '%test%'",
@@ -543,7 +543,7 @@ impl TestDatabaseManager {
sqlx::query(query)
.execute(&self.pool)
.await
.map_err(|e| TliError::DatabaseError(format!("Cleanup failed: {}", e)))?;
.map_err(|e| FxtError::DatabaseError(format!("Cleanup failed: {}", e)))?;
}
}
@@ -558,13 +558,13 @@ pub struct TestDatabase {
impl TestDatabase {
/// Create new test database
pub async fn new() -> TliResult<Self> {
pub async fn new() -> FxtResult<Self> {
let manager = TestDatabaseManager::new().await?;
Ok(Self { manager })
}
/// Verify order was persisted
pub async fn verify_order_persisted(&self, order_id: &str) -> TliResult<bool> {
pub async fn verify_order_persisted(&self, order_id: &str) -> FxtResult<bool> {
self.manager.verify_order_persisted(order_id).await
}
}
@@ -587,7 +587,7 @@ pub struct MarketDataPoint {
impl TestDataProvider {
/// Create new test data provider
pub async fn new() -> TliResult<Self> {
pub async fn new() -> FxtResult<Self> {
let mut provider = Self {
historical_data: HashMap::new(),
};
@@ -648,7 +648,7 @@ pub struct MockMLModel {
impl MLTestInfrastructure {
/// Create new ML testing infrastructure
pub async fn new() -> TliResult<Self> {
pub async fn new() -> FxtResult<Self> {
let mut models = HashMap::new();
// Add mock models with realistic characteristics