use std::io::{self, Read, Write};
use std::ops::{Deref, DerefMut};
use melin_app::{Application, ApplyCtx, RejectReason as TransportRejectReason};
use melin_exchange_core::exchange::Exchange;
use melin_exchange_core::snapshot as engine_snapshot;
use melin_trading::trading_event::TradingEvent;
use melin_types::types::{
AccountId, ExecutionReport, OrderId, QueryResponse, RejectReason as EngineRejectReason, Symbol,
};
const _: () = assert!(size_of::<melin_transport_core::pipeline::InputSlot<TradingEvent>>() == 128);
#[cfg(not(feature = "latency-trace"))]
const _: () = assert!(
size_of::<melin_transport_core::pipeline::OutputSlot<ExecutionReport, QueryResponse>>() == 424
);
const _: () = assert!(size_of::<melin_journal::JournalEvent<TradingEvent>>() == 64);
const _: () = assert!(size_of::<ExecutionReport>() == 64);
pub struct ServerApp(pub Exchange);
impl ServerApp {
pub fn new() -> Self {
ServerApp(Exchange::new())
}
}
impl Default for ServerApp {
fn default() -> Self {
Self::new()
}
}
impl Deref for ServerApp {
type Target = Exchange;
#[inline]
fn deref(&self) -> &Exchange {
&self.0
}
}
impl DerefMut for ServerApp {
#[inline]
fn deref_mut(&mut self) -> &mut Exchange {
&mut self.0
}
}
impl Application for ServerApp {
type Event = TradingEvent;
type Report = ExecutionReport;
type QueryResponse = QueryResponse;
const APP_VERSION: u16 = engine_snapshot::PAYLOAD_VERSION;
#[inline]
fn apply(
&mut self,
event: Self::Event,
ctx: &ApplyCtx,
out: &mut Vec<Self::Report>,
) -> Option<Self::QueryResponse> {
self.0.set_current_event_ts_ns(ctx.now_ns);
match event {
TradingEvent::AddInstrument { spec } => {
self.0.add_instrument(spec);
None
}
TradingEvent::Deposit {
account,
currency,
amount,
} => {
self.0.deposit(account, currency, amount);
None
}
TradingEvent::SubmitOrder { symbol, order } => {
self.0.execute(symbol, order, out);
None
}
TradingEvent::CancelOrder {
symbol,
account,
order_id,
} => {
self.0.cancel(symbol, account, order_id, out);
None
}
TradingEvent::SetRiskLimits { symbol, limits } => {
self.0.set_risk_limits(symbol, limits);
None
}
TradingEvent::CancelAll { account } => {
self.0.cancel_all(account, out);
None
}
TradingEvent::SetCircuitBreaker { symbol, config } => {
self.0.set_circuit_breaker(symbol, config);
None
}
TradingEvent::CancelReplace {
symbol,
account,
order_id,
new_price,
new_quantity,
} => {
self.0
.cancel_replace(symbol, account, order_id, new_price, new_quantity, out);
None
}
TradingEvent::SetFeeSchedule { symbol, schedule } => {
self.0.set_fee_schedule(symbol, schedule, out);
None
}
TradingEvent::ProvisionAccount { account, amount } => {
self.0.provision_account(account, amount);
None
}
TradingEvent::Withdraw {
account,
currency,
amount,
} => {
if let Err(reason) = self.0.withdraw(account, currency, amount) {
out.push(ExecutionReport::Rejected {
order_id: OrderId(0),
symbol: Symbol(0),
account,
reason,
});
}
None
}
TradingEvent::EndOfDay => {
self.0.end_of_day(out);
None
}
TradingEvent::DisableInstrument { symbol } => {
self.0.disable_instrument(symbol, out);
None
}
TradingEvent::EnableInstrument { symbol } => {
self.0.enable_instrument(symbol, out);
None
}
TradingEvent::RemoveInstrument { symbol } => {
self.0.remove_instrument(symbol, out);
None
}
TradingEvent::QueryStats => {
Some(QueryResponse::Stats {
active_connections: ctx.active_connections,
events_processed: ctx.events_processed,
journal_sequence: ctx.journal_sequence.get(),
})
}
TradingEvent::QueryPosition { account } => {
let (balances, count) = self.0.accounts().balances_for(account);
Some(QueryResponse::Position {
account,
balances,
count,
})
}
TradingEvent::QueryRequestSeq => {
Some(QueryResponse::RequestSeqHwm {
hwm: self.0.request_seq_hwm(ctx.key_hash),
})
}
}
}
#[inline]
fn tick(&mut self, now_ns: u64, out: &mut Vec<Self::Report>) {
self.0.drain_due_scheduled_tasks(now_ns, out);
}
#[inline]
fn check_request_seq(&mut self, key_hash: u64, seq: u64) -> bool {
Exchange::check_request_seq(&mut self.0, key_hash, seq)
}
fn prefault(&mut self) {
Exchange::prefault(&mut self.0);
}
fn clone_via_snapshot(&self) -> std::io::Result<Self> {
Ok(ServerApp(Exchange::clone_via_snapshot(&self.0)))
}
fn build_reject(event: &Self::Event, reason: TransportRejectReason) -> Self::Report {
let engine_reason = match reason {
TransportRejectReason::DuplicateRequest => EngineRejectReason::DuplicateRequest,
TransportRejectReason::ReplicaDisconnected => EngineRejectReason::ReplicaDisconnected,
TransportRejectReason::Superseded => EngineRejectReason::Superseded,
};
ExecutionReport::Rejected {
order_id: extract_order_id(event),
symbol: extract_symbol(event),
account: extract_account_id(event),
reason: engine_reason,
}
}
fn snapshot<W: Write>(&self, w: &mut W) -> io::Result<()> {
let bytes = engine_snapshot::encode_exchange_payload(&self.0);
w.write_all(&bytes)
}
fn restore<R: Read>(r: &mut R) -> io::Result<Self> {
let mut bytes = Vec::new();
r.read_to_end(&mut bytes)?;
engine_snapshot::decode_exchange_payload(&bytes)
.map(ServerApp)
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
}
}
fn extract_order_id(event: &TradingEvent) -> OrderId {
match event {
TradingEvent::SubmitOrder { order, .. } => order.id,
TradingEvent::CancelOrder { order_id, .. }
| TradingEvent::CancelReplace { order_id, .. } => *order_id,
_ => OrderId(0),
}
}
fn extract_account_id(event: &TradingEvent) -> AccountId {
match event {
TradingEvent::SubmitOrder { order, .. } => order.account,
TradingEvent::CancelOrder { account, .. }
| TradingEvent::CancelAll { account }
| TradingEvent::CancelReplace { account, .. }
| TradingEvent::Deposit { account, .. }
| TradingEvent::Withdraw { account, .. }
| TradingEvent::ProvisionAccount { account, .. }
| TradingEvent::QueryPosition { account } => *account,
_ => AccountId(0),
}
}
fn extract_symbol(event: &TradingEvent) -> Symbol {
match event {
TradingEvent::SubmitOrder { symbol, .. }
| TradingEvent::CancelOrder { symbol, .. }
| TradingEvent::CancelReplace { symbol, .. }
| TradingEvent::SetRiskLimits { symbol, .. }
| TradingEvent::SetCircuitBreaker { symbol, .. }
| TradingEvent::SetFeeSchedule { symbol, .. }
| TradingEvent::DisableInstrument { symbol }
| TradingEvent::EnableInstrument { symbol }
| TradingEvent::RemoveInstrument { symbol } => *symbol,
_ => Symbol(0),
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Cursor;
use std::num::NonZeroU64;
use melin_types::types::{
CurrencyId, InstrumentSpec, Order, OrderType, Price, Quantity, SelfTradeProtection, Side,
TimeInForce,
};
fn price(p: u64) -> Price {
Price(NonZeroU64::new(p).unwrap())
}
fn qty(q: u64) -> Quantity {
Quantity(NonZeroU64::new(q).unwrap())
}
fn seeded_app() -> ServerApp {
let mut ex = Exchange::new();
ex.add_instrument(InstrumentSpec {
symbol: Symbol(1),
base: CurrencyId(1),
quote: CurrencyId(2),
});
ex.deposit(AccountId(1), CurrencyId(2), 1_000_000);
ServerApp(ex)
}
#[test]
fn apply_submit_order_produces_placed_report() {
let mut app = seeded_app();
let mut reports = Vec::new();
let ctx = ApplyCtx {
now_ns: 0,
journal_sequence: melin_app::WireSeq::new(0),
active_connections: 0,
events_processed: 0,
key_hash: 0,
};
let ev = TradingEvent::SubmitOrder {
symbol: Symbol(1),
order: Order {
id: OrderId(1),
account: AccountId(1),
side: Side::Buy,
order_type: OrderType::Limit {
price: price(100),
post_only: false,
},
quantity: qty(10),
time_in_force: TimeInForce::GTC,
stp: SelfTradeProtection::Allow,
expiry_ns: 0,
},
};
<ServerApp as Application>::apply(&mut app, ev, &ctx, &mut reports);
assert!(
!reports.is_empty(),
"apply should emit at least one report for a resting order"
);
}
#[test]
fn tick_advances_scheduler_clock() {
let mut app = ServerApp(Exchange::new());
let mut reports = Vec::new();
<ServerApp as Application>::tick(&mut app, 1_000_000_000, &mut reports);
assert!(reports.is_empty());
}
#[test]
fn apply_query_request_seq_returns_per_key_hwm() {
let mut app = seeded_app();
let key_a: u64 = 0xAAAA_AAAA_AAAA_AAAA;
let key_b: u64 = 0xBBBB_BBBB_BBBB_BBBB;
for seq in 1..=7 {
assert!(<ServerApp as Application>::check_request_seq(
&mut app, key_a, seq
));
}
for seq in 1..=3 {
assert!(<ServerApp as Application>::check_request_seq(
&mut app, key_b, seq
));
}
let mut reports = Vec::new();
let mk_ctx = |kh| ApplyCtx {
now_ns: 0,
journal_sequence: melin_app::WireSeq::new(0),
active_connections: 0,
events_processed: 0,
key_hash: kh,
};
let resp_a = <ServerApp as Application>::apply(
&mut app,
TradingEvent::QueryRequestSeq,
&mk_ctx(key_a),
&mut reports,
);
assert_eq!(resp_a, Some(QueryResponse::RequestSeqHwm { hwm: 7 }));
let resp_b = <ServerApp as Application>::apply(
&mut app,
TradingEvent::QueryRequestSeq,
&mk_ctx(key_b),
&mut reports,
);
assert_eq!(resp_b, Some(QueryResponse::RequestSeqHwm { hwm: 3 }));
let resp_unknown = <ServerApp as Application>::apply(
&mut app,
TradingEvent::QueryRequestSeq,
&mk_ctx(0xDEAD_BEEF),
&mut reports,
);
assert_eq!(resp_unknown, Some(QueryResponse::RequestSeqHwm { hwm: 0 }));
assert_eq!(app.0.request_seq_hwm(key_a), 7);
assert_eq!(app.0.request_seq_hwm(key_b), 3);
}
#[test]
fn check_request_seq_rejects_duplicates() {
let mut app = ServerApp(Exchange::new());
assert!(<ServerApp as Application>::check_request_seq(
&mut app, 42, 1
));
assert!(<ServerApp as Application>::check_request_seq(
&mut app, 42, 2
));
assert!(!<ServerApp as Application>::check_request_seq(
&mut app, 42, 2
));
assert!(!<ServerApp as Application>::check_request_seq(
&mut app, 42, 1
));
}
#[test]
fn build_reject_maps_transport_reasons() {
let ev = TradingEvent::SubmitOrder {
symbol: Symbol(7),
order: Order {
id: OrderId(42),
account: AccountId(3),
side: Side::Buy,
order_type: OrderType::Market,
quantity: qty(1),
time_in_force: TimeInForce::IOC,
stp: SelfTradeProtection::Allow,
expiry_ns: 0,
},
};
let r =
<ServerApp as Application>::build_reject(&ev, TransportRejectReason::DuplicateRequest);
match r {
ExecutionReport::Rejected {
order_id,
symbol,
account,
reason,
} => {
assert_eq!(order_id, OrderId(42));
assert_eq!(symbol, Symbol(7));
assert_eq!(account, AccountId(3));
assert_eq!(reason, EngineRejectReason::DuplicateRequest);
}
other => panic!("expected Rejected, got {other:?}"),
}
let r = <ServerApp as Application>::build_reject(
&TradingEvent::CancelAll {
account: AccountId(9),
},
TransportRejectReason::ReplicaDisconnected,
);
match r {
ExecutionReport::Rejected {
order_id,
symbol,
account,
reason,
} => {
assert_eq!(order_id, OrderId(0));
assert_eq!(symbol, Symbol(0));
assert_eq!(account, AccountId(9));
assert_eq!(reason, EngineRejectReason::ReplicaDisconnected);
}
other => panic!("expected Rejected, got {other:?}"),
}
let r = <ServerApp as Application>::build_reject(
&TradingEvent::CancelAll {
account: AccountId(9),
},
TransportRejectReason::Superseded,
);
match r {
ExecutionReport::Rejected {
account, reason, ..
} => {
assert_eq!(account, AccountId(9));
assert_eq!(reason, EngineRejectReason::Superseded);
}
other => panic!("expected Rejected, got {other:?}"),
}
}
#[test]
fn apply_withdraw_emits_rejection_on_failure() {
let mut app = seeded_app();
let ctx = ApplyCtx {
now_ns: 0,
journal_sequence: melin_app::WireSeq::new(0),
active_connections: 0,
events_processed: 0,
key_hash: 0,
};
let mut reports = Vec::new();
<ServerApp as Application>::apply(
&mut app,
TradingEvent::Withdraw {
account: AccountId(1),
currency: CurrencyId(2),
amount: 2_000_000,
},
&ctx,
&mut reports,
);
assert_eq!(reports.len(), 1);
match reports[0] {
ExecutionReport::Rejected {
order_id,
symbol,
account,
reason,
} => {
assert_eq!(order_id, OrderId(0));
assert_eq!(symbol, Symbol(0));
assert_eq!(account, AccountId(1));
assert_eq!(reason, EngineRejectReason::InsufficientBalance);
}
ref other => panic!("expected Rejected, got {other:?}"),
}
let mut reports = Vec::new();
<ServerApp as Application>::apply(
&mut app,
TradingEvent::Withdraw {
account: AccountId(999),
currency: CurrencyId(2),
amount: 1,
},
&ctx,
&mut reports,
);
assert_eq!(reports.len(), 1);
match reports[0] {
ExecutionReport::Rejected {
reason, account, ..
} => {
assert_eq!(account, AccountId(999));
assert_eq!(reason, EngineRejectReason::UnknownAccount);
}
ref other => panic!("expected Rejected, got {other:?}"),
}
let mut placed = Vec::new();
<ServerApp as Application>::apply(
&mut app,
TradingEvent::SubmitOrder {
symbol: Symbol(1),
order: Order {
id: OrderId(1),
account: AccountId(1),
side: Side::Buy,
order_type: OrderType::Limit {
price: price(100),
post_only: false,
},
quantity: qty(10),
time_in_force: TimeInForce::GTC,
stp: SelfTradeProtection::Allow,
expiry_ns: 0,
},
},
&ctx,
&mut placed,
);
let mut reports = Vec::new();
<ServerApp as Application>::apply(
&mut app,
TradingEvent::Withdraw {
account: AccountId(1),
currency: CurrencyId(2),
amount: 1,
},
&ctx,
&mut reports,
);
assert_eq!(reports.len(), 1);
match reports[0] {
ExecutionReport::Rejected {
reason, account, ..
} => {
assert_eq!(account, AccountId(1));
assert_eq!(reason, EngineRejectReason::HasRestingOrders);
}
ref other => panic!("expected Rejected, got {other:?}"),
}
let mut reports = Vec::new();
let mut clean = ServerApp::new();
clean.0.deposit(AccountId(7), CurrencyId(2), 500);
<ServerApp as Application>::apply(
&mut clean,
TradingEvent::Withdraw {
account: AccountId(7),
currency: CurrencyId(2),
amount: 200,
},
&ctx,
&mut reports,
);
assert!(
reports.is_empty(),
"successful withdraw must not emit reports"
);
}
#[test]
fn snapshot_restore_round_trip_preserves_state() {
let mut before = seeded_app();
let mut reports = Vec::new();
before.0.execute(
Symbol(1),
Order {
id: OrderId(1),
account: AccountId(1),
side: Side::Buy,
order_type: OrderType::Limit {
price: price(100),
post_only: false,
},
quantity: qty(10),
time_in_force: TimeInForce::GTC,
stp: SelfTradeProtection::Allow,
expiry_ns: 0,
},
&mut reports,
);
let reports_before = reports.clone();
let mut buf = Vec::new();
<ServerApp as Application>::snapshot(&before, &mut buf).expect("snapshot");
let mut cursor = Cursor::new(buf);
let mut after = <ServerApp as Application>::restore(&mut cursor).expect("restore");
let mut reports_after = reports_before.clone();
reports_after.clear();
after.0.execute(
Symbol(1),
Order {
id: OrderId(2),
account: AccountId(1),
side: Side::Buy,
order_type: OrderType::Limit {
price: price(99),
post_only: false,
},
quantity: qty(5),
time_in_force: TimeInForce::GTC,
stp: SelfTradeProtection::Allow,
expiry_ns: 0,
},
&mut reports_after,
);
assert!(
!reports_after.is_empty(),
"restored exchange must accept orders"
);
}
}