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

feat: add broker_gateway_service to workspace, fix compilation

Browse Source 
Add the 8.4k-line broker gateway (AMP Futures/CQG FIX routing) to
workspace members. Fix 16 compilation errors from API drift:
- Replace sqlx::query! macros with runtime sqlx::query (no .sqlx cache)
- Add FromRow structs for typed query results
- Remove unused imports, use safe indexing

7 tests pass.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-22 02:50:45 +01:00
parent f187ce1dd4
commit 704ce8eff6
4 changed files with 146 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
Cargo.lock generated
View File

@@ -1806,6 +1806,52 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "broker_gateway_service"
version = "1.0.0"
dependencies = [
"anyhow",
"async-stream",
"async-trait",
"axum 0.7.9",
"base64 0.22.1",
"bigdecimal",
"bytes",
"chrono",
"common",
"config",
"criterion",
"futures",
"http-body-util",
"hyper 1.7.0",
"hyper-util",
"num-traits",
"once_cell",
"prometheus",
"prost 0.14.1",
"prost-build",
"redis",
"rust_decimal",
"serde",
"serde_json",
"serial_test",
"sha2",
"sqlx",
"thiserror 1.0.69",
"tokio",
"tokio-stream",
"tokio-test",
"tonic",
"tonic-health",
"tonic-prost",
"tonic-prost-build",
"tonic-reflection",
"tower 0.4.13",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
name = "bstr"
version = "1.12.0"

1
Cargo.toml
View File

@@ -122,6 +122,7 @@ members = [
"database",
"config",
"services/backtesting_service",
"services/broker_gateway_service",
"services/trading_service",
"services/ml_training_service",
"services/data_acquisition_service",

110
services/broker_gateway_service/src/recovery/mod.rs
View File

@@ -43,7 +43,7 @@ impl SessionRecovery {
info!("Recovering FIX session: {}", self.session_id);
// Query database for persisted session state
let session = sqlx::query!(
let session: Option<SessionInfoRow> = sqlx::query_as(
r#"
SELECT
sender_seq_num,
@@ -57,8 +57,8 @@ impl SessionRecovery {
ORDER BY updated_at DESC
LIMIT 1
"#,
self.session_id
)
.bind(&self.session_id)
.fetch_optional(&self.db_pool)
.await?;
@@ -103,22 +103,22 @@ impl SessionRecovery {
// Parse session_id: "FOXHUNT_CLIENT-CQG"
let parts: Vec<&str> = self.session_id.split('-').collect();
let (sender_comp_id, target_comp_id) = if parts.len() == 2 {
(parts[0], parts[1])
(parts.first().copied().unwrap_or("FOXHUNT_CLIENT"), parts.get(1).copied().unwrap_or("CQG"))
} else {
("FOXHUNT_CLIENT", "CQG")
};
sqlx::query!(
sqlx::query(
r#"
INSERT INTO broker_sessions
(session_id, sender_comp_id, target_comp_id, sender_seq_num, target_seq_num, session_state, created_at, updated_at)
VALUES ($1, $2, $3, 1, 1, 'DISCONNECTED', NOW(), NOW())
ON CONFLICT (session_id) DO NOTHING
"#,
self.session_id,
sender_comp_id,
target_comp_id
)
.bind(&self.session_id)
.bind(sender_comp_id)
.bind(target_comp_id)
.execute(&self.db_pool)
.await?;
@@ -135,14 +135,14 @@ impl SessionRecovery {
*state = SessionState::LoggingIn;
}
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_sessions
SET session_state = 'RECONNECTING', updated_at = NOW()
WHERE session_id = $1
"#,
self.session_id
)
.bind(&self.session_id)
.execute(&self.db_pool)
.await?;
@@ -155,14 +155,14 @@ impl SessionRecovery {
*state = SessionState::Active;
}
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_sessions
SET session_state = 'ACTIVE', connected_at = NOW(), updated_at = NOW()
WHERE session_id = $1
"#,
self.session_id
)
.bind(&self.session_id)
.execute(&self.db_pool)
.await?;
@@ -176,16 +176,16 @@ impl SessionRecovery {
sender_seq_num: i64,
target_seq_num: i64,
) -> anyhow::Result<()> {
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_sessions
SET sender_seq_num = $2, target_seq_num = $3, updated_at = NOW()
WHERE session_id = $1
"#,
self.session_id,
sender_seq_num,
target_seq_num
)
.bind(&self.session_id)
.bind(sender_seq_num)
.bind(target_seq_num)
.execute(&self.db_pool)
.await?;
@@ -201,14 +201,14 @@ impl SessionRecovery {
*state = SessionState::Disconnected;
}
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_sessions
SET session_state = 'DISCONNECTED', disconnected_at = NOW(), updated_at = NOW()
WHERE session_id = $1
"#,
self.session_id
)
.bind(&self.session_id)
.execute(&self.db_pool)
.await?;
@@ -216,6 +216,17 @@ impl SessionRecovery {
}
}
/// Row type for session info query
#[derive(sqlx::FromRow)]
struct SessionInfoRow {
sender_seq_num: i64,
target_seq_num: i64,
session_state: String,
last_heartbeat_sent: Option<chrono::DateTime<chrono::Utc>>,
last_heartbeat_received: Option<chrono::DateTime<chrono::Utc>>,
connected_at: Option<chrono::DateTime<chrono::Utc>>,
}
/// Session information recovered from database
#[derive(Debug, Clone)]
pub struct SessionInfo {
@@ -239,6 +250,21 @@ pub struct OrderRecovery {
db_pool: PgPool,
}
/// Row type for unsent order query
#[derive(sqlx::FromRow)]
struct UnsentOrderRow {
client_order_id: String,
account_id: String,
symbol: String,
side: String,
order_type: String,
quantity: rust_decimal::Decimal,
price: Option<rust_decimal::Decimal>,
stop_price: Option<rust_decimal::Decimal>,
metadata: Option<serde_json::Value>,
submitted_at: Option<chrono::DateTime<chrono::Utc>>,
}
impl OrderRecovery {
/// Create a new order recovery manager
pub fn new(db_pool: PgPool) -> Self {
@@ -249,7 +275,7 @@ impl OrderRecovery {
pub async fn recover_unsent_orders(&self) -> anyhow::Result<Vec<UnsentOrder>> {
info!("Recovering unsent orders from database");
let orders = sqlx::query!(
let orders: Vec<UnsentOrderRow> = sqlx::query_as(
r#"
SELECT
client_order_id,
@@ -265,7 +291,7 @@ impl OrderRecovery {
FROM broker_orders
WHERE status = 'PENDING_SUBMIT'
ORDER BY created_at ASC
"#
"#,
)
.fetch_all(&self.db_pool)
.await?;
@@ -296,14 +322,14 @@ impl OrderRecovery {
/// Mark order as submitted after successful send
pub async fn mark_order_submitted(&self, client_order_id: &str) -> anyhow::Result<()> {
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_orders
SET status = 'SUBMITTED', updated_at = NOW()
WHERE client_order_id = $1
"#,
client_order_id
)
.bind(client_order_id)
.execute(&self.db_pool)
.await?;
@@ -315,9 +341,9 @@ impl OrderRecovery {
pub async fn mark_order_failed(
&self,
client_order_id: &str,
error: &str,
error_msg: &str,
) -> anyhow::Result<()> {
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_orders
SET status = 'REJECTED',
@@ -329,13 +355,13 @@ impl OrderRecovery {
updated_at = NOW()
WHERE client_order_id = $1
"#,
client_order_id,
error
)
.bind(client_order_id)
.bind(error_msg)
.execute(&self.db_pool)
.await?;
error!("Order {} marked as REJECTED: {}", client_order_id, error);
error!("Order {} marked as REJECTED: {}", client_order_id, error_msg);
Ok(())
}
}
@@ -371,6 +397,18 @@ pub struct PositionRecovery {
db_pool: PgPool,
}
/// Row type for position query
#[derive(sqlx::FromRow)]
struct PositionRow {
symbol: String,
quantity: rust_decimal::Decimal,
avg_entry_price: Option<rust_decimal::Decimal>,
market_value: Option<rust_decimal::Decimal>,
unrealized_pnl: Option<rust_decimal::Decimal>,
realized_pnl: Option<rust_decimal::Decimal>,
last_updated: chrono::DateTime<chrono::Utc>,
}
impl PositionRecovery {
/// Create a new position recovery manager
pub fn new(db_pool: PgPool) -> Self {
@@ -381,7 +419,7 @@ impl PositionRecovery {
pub async fn reconcile_positions(&self, account_id: &str) -> anyhow::Result<Vec<Position>> {
info!("Reconciling positions for account: {}", account_id);
let positions = sqlx::query!(
let positions: Vec<PositionRow> = sqlx::query_as(
r#"
SELECT
symbol,
@@ -394,8 +432,8 @@ impl PositionRecovery {
FROM broker_positions
WHERE account_id = $1
"#,
account_id
)
.bind(account_id)
.fetch_all(&self.db_pool)
.await?;
@@ -423,7 +461,7 @@ impl PositionRecovery {
/// Update position in database
pub async fn update_position(&self, account_id: &str, position: &Position) -> anyhow::Result<()> {
sqlx::query!(
sqlx::query(
r#"
INSERT INTO broker_positions
(account_id, symbol, quantity, avg_entry_price, market_value, unrealized_pnl, realized_pnl, last_updated, created_at)
@@ -436,14 +474,14 @@ impl PositionRecovery {
realized_pnl = EXCLUDED.realized_pnl,
last_updated = NOW()
"#,
account_id,
position.symbol,
position.quantity,
position.avg_entry_price,
position.market_value,
position.unrealized_pnl,
position.realized_pnl
)
.bind(account_id)
.bind(&position.symbol)
.bind(position.quantity)
.bind(position.avg_entry_price)
.bind(position.market_value)
.bind(position.unrealized_pnl)
.bind(position.realized_pnl)
.execute(&self.db_pool)
.await?;

47
services/broker_gateway_service/src/service.rs
View File

@@ -5,8 +5,6 @@
use sqlx::PgPool;
use std::sync::Arc;
use std::str::FromStr;
use rust_decimal::Decimal;
use tokio::sync::RwLock;
use tonic::{Request, Response, Status};
use tracing::{error, info, instrument, warn};
@@ -139,24 +137,29 @@ impl broker_gateway_service_server::BrokerGatewayService for BrokerGatewayServic
_ => return Err(Status::invalid_argument("Invalid order type")),
};
sqlx::query!(
let quantity_decimal = rust_decimal::Decimal::try_from(req.quantity).ok();
let price_decimal = req.price.and_then(|p| rust_decimal::Decimal::try_from(p).ok());
let stop_price_decimal = req.stop_price.and_then(|p| rust_decimal::Decimal::try_from(p).ok());
let metadata_json = serde_json::to_value(&req.metadata).ok();
sqlx::query(
r#"
INSERT INTO broker_orders
(client_order_id, account_id, symbol, side, order_type, quantity, price, stop_price, status, metadata, submitted_at, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW(), NOW())
"#,
client_order_id,
req.account_id,
req.symbol,
side_str,
order_type_str,
Decimal::from_str(&req.quantity.to_string()).ok(),
req.price.and_then(|p| Decimal::from_str(&p.to_string()).ok()),
req.stop_price.and_then(|p| Decimal::from_str(&p.to_string()).ok()),
"PENDING_SUBMIT",
serde_json::to_value(&req.metadata).ok(),
submitted_at,
)
.bind(&client_order_id)
.bind(&req.account_id)
.bind(&req.symbol)
.bind(side_str)
.bind(order_type_str)
.bind(quantity_decimal)
.bind(price_decimal)
.bind(stop_price_decimal)
.bind("PENDING_SUBMIT")
.bind(metadata_json)
.bind(submitted_at)
.execute(&self.db_pool)
.await
.map_err(|e| {
@@ -206,15 +209,15 @@ impl broker_gateway_service_server::BrokerGatewayService for BrokerGatewayServic
);
// Fetch order from database
let order = sqlx::query!(
let order_row: Option<(String,)> = sqlx::query_as(
r#"
SELECT status
FROM broker_orders
WHERE client_order_id = $1 AND account_id = $2
"#,
req.client_order_id,
req.account_id
)
.bind(&req.client_order_id)
.bind(&req.account_id)
.fetch_optional(&self.db_pool)
.await
.map_err(|e| {
@@ -222,28 +225,28 @@ impl broker_gateway_service_server::BrokerGatewayService for BrokerGatewayServic
Status::internal(format!("Database error: {}", e))
})?;
let order = order.ok_or_else(|| Status::not_found("Order not found"))?;
let (status,) = order_row.ok_or_else(|| Status::not_found("Order not found"))?;
// Check if order is cancellable
if !matches!(
order.status.as_str(),
status.as_str(),
"PENDING_SUBMIT" | "SUBMITTED" | "PARTIALLY_FILLED"
) {
return Err(Status::failed_precondition(format!(
"Order cannot be cancelled (status: {})",
order.status
status
)));
}
// MVP: Update status to CANCEL_PENDING (no actual FIX send)
sqlx::query!(
sqlx::query(
r#"
UPDATE broker_orders
SET status = 'CANCEL_PENDING', updated_at = NOW()
WHERE client_order_id = $1
"#,
req.client_order_id
)
.bind(&req.client_order_id)
.execute(&self.db_pool)
.await
.map_err(|e| {