1use std::io::{self, Read, Write};
15use std::ops::{Deref, DerefMut};
16
17use melin_app::{Application, ApplyCtx, RejectReason as TransportRejectReason};
18use melin_exchange_core::exchange::Exchange;
19use melin_exchange_core::snapshot as engine_snapshot;
20use melin_trading::trading_event::TradingEvent;
21use melin_types::types::{
22 AccountId, ExecutionReport, OrderId, QueryResponse, RejectReason as EngineRejectReason, Symbol,
23};
24
25const _: () = assert!(size_of::<melin_transport_core::pipeline::InputSlot<TradingEvent>>() == 128);
39#[cfg(not(feature = "latency-trace"))]
45const _: () = assert!(
46 size_of::<melin_transport_core::pipeline::OutputSlot<ExecutionReport, QueryResponse>>() == 424
47);
48const _: () = assert!(size_of::<melin_journal::JournalEvent<TradingEvent>>() == 64);
49const _: () = assert!(size_of::<ExecutionReport>() == 64);
50
51pub struct ServerApp(pub Exchange);
62
63impl ServerApp {
64 pub fn new() -> Self {
68 ServerApp(Exchange::new())
69 }
70}
71
72impl Default for ServerApp {
73 fn default() -> Self {
74 Self::new()
75 }
76}
77
78impl Deref for ServerApp {
79 type Target = Exchange;
80
81 #[inline]
82 fn deref(&self) -> &Exchange {
83 &self.0
84 }
85}
86
87impl DerefMut for ServerApp {
88 #[inline]
89 fn deref_mut(&mut self) -> &mut Exchange {
90 &mut self.0
91 }
92}
93
94impl Application for ServerApp {
95 type Event = TradingEvent;
96 type Report = ExecutionReport;
97 type QueryResponse = QueryResponse;
98
99 const APP_VERSION: u16 = engine_snapshot::PAYLOAD_VERSION;
103
104 #[inline]
109 fn apply(
110 &mut self,
111 event: Self::Event,
112 ctx: &ApplyCtx,
113 out: &mut Vec<Self::Report>,
114 ) -> Option<Self::QueryResponse> {
115 self.0.set_current_event_ts_ns(ctx.now_ns);
121 match event {
122 TradingEvent::AddInstrument { spec } => {
123 self.0.add_instrument(spec);
124 None
125 }
126 TradingEvent::Deposit {
127 account,
128 currency,
129 amount,
130 } => {
131 self.0.deposit(account, currency, amount);
132 None
133 }
134 TradingEvent::SubmitOrder { symbol, order } => {
135 self.0.execute(symbol, order, out);
136 None
137 }
138 TradingEvent::CancelOrder {
139 symbol,
140 account,
141 order_id,
142 } => {
143 self.0.cancel(symbol, account, order_id, out);
144 None
145 }
146 TradingEvent::SetRiskLimits { symbol, limits } => {
147 self.0.set_risk_limits(symbol, limits);
148 None
149 }
150 TradingEvent::CancelAll { account } => {
151 self.0.cancel_all(account, out);
152 None
153 }
154 TradingEvent::SetCircuitBreaker { symbol, config } => {
155 self.0.set_circuit_breaker(symbol, config);
156 None
157 }
158 TradingEvent::CancelReplace {
159 symbol,
160 account,
161 order_id,
162 new_price,
163 new_quantity,
164 } => {
165 self.0
166 .cancel_replace(symbol, account, order_id, new_price, new_quantity, out);
167 None
168 }
169 TradingEvent::SetFeeSchedule { symbol, schedule } => {
170 self.0.set_fee_schedule(symbol, schedule, out);
171 None
172 }
173 TradingEvent::ProvisionAccount { account, amount } => {
174 self.0.provision_account(account, amount);
175 None
176 }
177 TradingEvent::Withdraw {
178 account,
179 currency,
180 amount,
181 } => {
182 if let Err(reason) = self.0.withdraw(account, currency, amount) {
183 out.push(ExecutionReport::Rejected {
192 order_id: OrderId(0),
193 symbol: Symbol(0),
194 account,
195 reason,
196 });
197 }
198 None
199 }
200 TradingEvent::EndOfDay => {
201 self.0.end_of_day(out);
202 None
203 }
204 TradingEvent::DisableInstrument { symbol } => {
205 self.0.disable_instrument(symbol, out);
206 None
207 }
208 TradingEvent::EnableInstrument { symbol } => {
209 self.0.enable_instrument(symbol, out);
210 None
211 }
212 TradingEvent::RemoveInstrument { symbol } => {
213 self.0.remove_instrument(symbol, out);
214 None
215 }
216 TradingEvent::QueryStats => {
217 Some(QueryResponse::Stats {
221 active_connections: ctx.active_connections,
222 events_processed: ctx.events_processed,
223 journal_sequence: ctx.journal_sequence.get(),
224 })
225 }
226 TradingEvent::QueryPosition { account } => {
227 let (balances, count) = self.0.accounts().balances_for(account);
228 Some(QueryResponse::Position {
229 account,
230 balances,
231 count,
232 })
233 }
234 TradingEvent::QueryRequestSeq => {
235 Some(QueryResponse::RequestSeqHwm {
240 hwm: self.0.request_seq_hwm(ctx.key_hash),
241 })
242 }
243 }
244 }
245
246 #[inline]
247 fn tick(&mut self, now_ns: u64, out: &mut Vec<Self::Report>) {
248 self.0.drain_due_scheduled_tasks(now_ns, out);
249 }
250
251 #[inline]
252 fn check_request_seq(&mut self, key_hash: u64, seq: u64) -> bool {
253 Exchange::check_request_seq(&mut self.0, key_hash, seq)
254 }
255
256 fn prefault(&mut self) {
261 Exchange::prefault(&mut self.0);
262 }
263
264 fn clone_via_snapshot(&self) -> std::io::Result<Self> {
269 Ok(ServerApp(Exchange::clone_via_snapshot(&self.0)))
270 }
271
272 fn build_reject(event: &Self::Event, reason: TransportRejectReason) -> Self::Report {
273 let engine_reason = match reason {
274 TransportRejectReason::DuplicateRequest => EngineRejectReason::DuplicateRequest,
275 TransportRejectReason::ReplicaDisconnected => EngineRejectReason::ReplicaDisconnected,
276 TransportRejectReason::Superseded => EngineRejectReason::Superseded,
277 };
278 ExecutionReport::Rejected {
279 order_id: extract_order_id(event),
280 symbol: extract_symbol(event),
281 account: extract_account_id(event),
282 reason: engine_reason,
283 }
284 }
285
286 fn snapshot<W: Write>(&self, w: &mut W) -> io::Result<()> {
293 let bytes = engine_snapshot::encode_exchange_payload(&self.0);
294 w.write_all(&bytes)
295 }
296
297 fn restore<R: Read>(r: &mut R) -> io::Result<Self> {
298 let mut bytes = Vec::new();
299 r.read_to_end(&mut bytes)?;
300 engine_snapshot::decode_exchange_payload(&bytes)
301 .map(ServerApp)
302 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
303 }
304}
305
306fn extract_order_id(event: &TradingEvent) -> OrderId {
310 match event {
311 TradingEvent::SubmitOrder { order, .. } => order.id,
312 TradingEvent::CancelOrder { order_id, .. }
313 | TradingEvent::CancelReplace { order_id, .. } => *order_id,
314 _ => OrderId(0),
315 }
316}
317
318fn extract_account_id(event: &TradingEvent) -> AccountId {
319 match event {
320 TradingEvent::SubmitOrder { order, .. } => order.account,
321 TradingEvent::CancelOrder { account, .. }
322 | TradingEvent::CancelAll { account }
323 | TradingEvent::CancelReplace { account, .. }
324 | TradingEvent::Deposit { account, .. }
325 | TradingEvent::Withdraw { account, .. }
326 | TradingEvent::ProvisionAccount { account, .. }
327 | TradingEvent::QueryPosition { account } => *account,
328 _ => AccountId(0),
329 }
330}
331
332fn extract_symbol(event: &TradingEvent) -> Symbol {
333 match event {
334 TradingEvent::SubmitOrder { symbol, .. }
335 | TradingEvent::CancelOrder { symbol, .. }
336 | TradingEvent::CancelReplace { symbol, .. }
337 | TradingEvent::SetRiskLimits { symbol, .. }
338 | TradingEvent::SetCircuitBreaker { symbol, .. }
339 | TradingEvent::SetFeeSchedule { symbol, .. }
340 | TradingEvent::DisableInstrument { symbol }
341 | TradingEvent::EnableInstrument { symbol }
342 | TradingEvent::RemoveInstrument { symbol } => *symbol,
343 _ => Symbol(0),
344 }
345}
346
347#[cfg(test)]
348mod tests {
349 use super::*;
350
351 use std::io::Cursor;
352 use std::num::NonZeroU64;
353
354 use melin_types::types::{
355 CurrencyId, InstrumentSpec, Order, OrderType, Price, Quantity, SelfTradeProtection, Side,
356 TimeInForce,
357 };
358
359 fn price(p: u64) -> Price {
360 Price(NonZeroU64::new(p).unwrap())
361 }
362 fn qty(q: u64) -> Quantity {
363 Quantity(NonZeroU64::new(q).unwrap())
364 }
365
366 fn seeded_app() -> ServerApp {
369 let mut ex = Exchange::new();
370 ex.add_instrument(InstrumentSpec {
371 symbol: Symbol(1),
372 base: CurrencyId(1),
373 quote: CurrencyId(2),
374 });
375 ex.deposit(AccountId(1), CurrencyId(2), 1_000_000);
376 ServerApp(ex)
377 }
378
379 #[test]
380 fn apply_submit_order_produces_placed_report() {
381 let mut app = seeded_app();
382 let mut reports = Vec::new();
383 let ctx = ApplyCtx {
384 now_ns: 0,
385 journal_sequence: melin_app::WireSeq::new(0),
386 active_connections: 0,
387 events_processed: 0,
388 key_hash: 0,
389 };
390 let ev = TradingEvent::SubmitOrder {
391 symbol: Symbol(1),
392 order: Order {
393 id: OrderId(1),
394 account: AccountId(1),
395 side: Side::Buy,
396 order_type: OrderType::Limit {
397 price: price(100),
398 post_only: false,
399 },
400 quantity: qty(10),
401 time_in_force: TimeInForce::GTC,
402 stp: SelfTradeProtection::Allow,
403 expiry_ns: 0,
404 },
405 };
406 <ServerApp as Application>::apply(&mut app, ev, &ctx, &mut reports);
407 assert!(
408 !reports.is_empty(),
409 "apply should emit at least one report for a resting order"
410 );
411 }
412
413 #[test]
414 fn tick_advances_scheduler_clock() {
415 let mut app = ServerApp(Exchange::new());
419 let mut reports = Vec::new();
420 <ServerApp as Application>::tick(&mut app, 1_000_000_000, &mut reports);
421 assert!(reports.is_empty());
422 }
423
424 #[test]
425 fn apply_query_request_seq_returns_per_key_hwm() {
426 let mut app = seeded_app();
427
428 let key_a: u64 = 0xAAAA_AAAA_AAAA_AAAA;
431 let key_b: u64 = 0xBBBB_BBBB_BBBB_BBBB;
432 for seq in 1..=7 {
433 assert!(<ServerApp as Application>::check_request_seq(
434 &mut app, key_a, seq
435 ));
436 }
437 for seq in 1..=3 {
438 assert!(<ServerApp as Application>::check_request_seq(
439 &mut app, key_b, seq
440 ));
441 }
442
443 let mut reports = Vec::new();
444 let mk_ctx = |kh| ApplyCtx {
445 now_ns: 0,
446 journal_sequence: melin_app::WireSeq::new(0),
447 active_connections: 0,
448 events_processed: 0,
449 key_hash: kh,
450 };
451
452 let resp_a = <ServerApp as Application>::apply(
455 &mut app,
456 TradingEvent::QueryRequestSeq,
457 &mk_ctx(key_a),
458 &mut reports,
459 );
460 assert_eq!(resp_a, Some(QueryResponse::RequestSeqHwm { hwm: 7 }));
461
462 let resp_b = <ServerApp as Application>::apply(
463 &mut app,
464 TradingEvent::QueryRequestSeq,
465 &mk_ctx(key_b),
466 &mut reports,
467 );
468 assert_eq!(resp_b, Some(QueryResponse::RequestSeqHwm { hwm: 3 }));
469
470 let resp_unknown = <ServerApp as Application>::apply(
472 &mut app,
473 TradingEvent::QueryRequestSeq,
474 &mk_ctx(0xDEAD_BEEF),
475 &mut reports,
476 );
477 assert_eq!(resp_unknown, Some(QueryResponse::RequestSeqHwm { hwm: 0 }));
478
479 assert_eq!(app.0.request_seq_hwm(key_a), 7);
481 assert_eq!(app.0.request_seq_hwm(key_b), 3);
482 }
483
484 #[test]
485 fn check_request_seq_rejects_duplicates() {
486 let mut app = ServerApp(Exchange::new());
487 assert!(<ServerApp as Application>::check_request_seq(
488 &mut app, 42, 1
489 ));
490 assert!(<ServerApp as Application>::check_request_seq(
491 &mut app, 42, 2
492 ));
493 assert!(!<ServerApp as Application>::check_request_seq(
494 &mut app, 42, 2
495 ));
496 assert!(!<ServerApp as Application>::check_request_seq(
497 &mut app, 42, 1
498 ));
499 }
500
501 #[test]
502 fn build_reject_maps_transport_reasons() {
503 let ev = TradingEvent::SubmitOrder {
504 symbol: Symbol(7),
505 order: Order {
506 id: OrderId(42),
507 account: AccountId(3),
508 side: Side::Buy,
509 order_type: OrderType::Market,
510 quantity: qty(1),
511 time_in_force: TimeInForce::IOC,
512 stp: SelfTradeProtection::Allow,
513 expiry_ns: 0,
514 },
515 };
516 let r =
517 <ServerApp as Application>::build_reject(&ev, TransportRejectReason::DuplicateRequest);
518 match r {
519 ExecutionReport::Rejected {
520 order_id,
521 symbol,
522 account,
523 reason,
524 } => {
525 assert_eq!(order_id, OrderId(42));
526 assert_eq!(symbol, Symbol(7));
527 assert_eq!(account, AccountId(3));
528 assert_eq!(reason, EngineRejectReason::DuplicateRequest);
529 }
530 other => panic!("expected Rejected, got {other:?}"),
531 }
532
533 let r = <ServerApp as Application>::build_reject(
534 &TradingEvent::CancelAll {
535 account: AccountId(9),
536 },
537 TransportRejectReason::ReplicaDisconnected,
538 );
539 match r {
540 ExecutionReport::Rejected {
541 order_id,
542 symbol,
543 account,
544 reason,
545 } => {
546 assert_eq!(order_id, OrderId(0));
547 assert_eq!(symbol, Symbol(0));
548 assert_eq!(account, AccountId(9));
549 assert_eq!(reason, EngineRejectReason::ReplicaDisconnected);
550 }
551 other => panic!("expected Rejected, got {other:?}"),
552 }
553
554 let r = <ServerApp as Application>::build_reject(
555 &TradingEvent::CancelAll {
556 account: AccountId(9),
557 },
558 TransportRejectReason::Superseded,
559 );
560 match r {
561 ExecutionReport::Rejected {
562 account, reason, ..
563 } => {
564 assert_eq!(account, AccountId(9));
565 assert_eq!(reason, EngineRejectReason::Superseded);
566 }
567 other => panic!("expected Rejected, got {other:?}"),
568 }
569 }
570
571 #[test]
572 fn apply_withdraw_emits_rejection_on_failure() {
573 let mut app = seeded_app();
574 let ctx = ApplyCtx {
575 now_ns: 0,
576 journal_sequence: melin_app::WireSeq::new(0),
577 active_connections: 0,
578 events_processed: 0,
579 key_hash: 0,
580 };
581
582 let mut reports = Vec::new();
585 <ServerApp as Application>::apply(
586 &mut app,
587 TradingEvent::Withdraw {
588 account: AccountId(1),
589 currency: CurrencyId(2),
590 amount: 2_000_000,
591 },
592 &ctx,
593 &mut reports,
594 );
595 assert_eq!(reports.len(), 1);
596 match reports[0] {
597 ExecutionReport::Rejected {
598 order_id,
599 symbol,
600 account,
601 reason,
602 } => {
603 assert_eq!(order_id, OrderId(0));
604 assert_eq!(symbol, Symbol(0));
605 assert_eq!(account, AccountId(1));
606 assert_eq!(reason, EngineRejectReason::InsufficientBalance);
607 }
608 ref other => panic!("expected Rejected, got {other:?}"),
609 }
610
611 let mut reports = Vec::new();
614 <ServerApp as Application>::apply(
615 &mut app,
616 TradingEvent::Withdraw {
617 account: AccountId(999),
618 currency: CurrencyId(2),
619 amount: 1,
620 },
621 &ctx,
622 &mut reports,
623 );
624 assert_eq!(reports.len(), 1);
625 match reports[0] {
626 ExecutionReport::Rejected {
627 reason, account, ..
628 } => {
629 assert_eq!(account, AccountId(999));
630 assert_eq!(reason, EngineRejectReason::UnknownAccount);
631 }
632 ref other => panic!("expected Rejected, got {other:?}"),
633 }
634
635 let mut placed = Vec::new();
637 <ServerApp as Application>::apply(
638 &mut app,
639 TradingEvent::SubmitOrder {
640 symbol: Symbol(1),
641 order: Order {
642 id: OrderId(1),
643 account: AccountId(1),
644 side: Side::Buy,
645 order_type: OrderType::Limit {
646 price: price(100),
647 post_only: false,
648 },
649 quantity: qty(10),
650 time_in_force: TimeInForce::GTC,
651 stp: SelfTradeProtection::Allow,
652 expiry_ns: 0,
653 },
654 },
655 &ctx,
656 &mut placed,
657 );
658
659 let mut reports = Vec::new();
660 <ServerApp as Application>::apply(
661 &mut app,
662 TradingEvent::Withdraw {
663 account: AccountId(1),
664 currency: CurrencyId(2),
665 amount: 1,
666 },
667 &ctx,
668 &mut reports,
669 );
670 assert_eq!(reports.len(), 1);
671 match reports[0] {
672 ExecutionReport::Rejected {
673 reason, account, ..
674 } => {
675 assert_eq!(account, AccountId(1));
676 assert_eq!(reason, EngineRejectReason::HasRestingOrders);
677 }
678 ref other => panic!("expected Rejected, got {other:?}"),
679 }
680
681 let mut reports = Vec::new();
683 let mut clean = ServerApp::new();
684 clean.0.deposit(AccountId(7), CurrencyId(2), 500);
685 <ServerApp as Application>::apply(
686 &mut clean,
687 TradingEvent::Withdraw {
688 account: AccountId(7),
689 currency: CurrencyId(2),
690 amount: 200,
691 },
692 &ctx,
693 &mut reports,
694 );
695 assert!(
696 reports.is_empty(),
697 "successful withdraw must not emit reports"
698 );
699 }
700
701 #[test]
702 fn snapshot_restore_round_trip_preserves_state() {
703 let mut before = seeded_app();
704 let mut reports = Vec::new();
705 before.0.execute(
708 Symbol(1),
709 Order {
710 id: OrderId(1),
711 account: AccountId(1),
712 side: Side::Buy,
713 order_type: OrderType::Limit {
714 price: price(100),
715 post_only: false,
716 },
717 quantity: qty(10),
718 time_in_force: TimeInForce::GTC,
719 stp: SelfTradeProtection::Allow,
720 expiry_ns: 0,
721 },
722 &mut reports,
723 );
724 let reports_before = reports.clone();
725
726 let mut buf = Vec::new();
727 <ServerApp as Application>::snapshot(&before, &mut buf).expect("snapshot");
728
729 let mut cursor = Cursor::new(buf);
730 let mut after = <ServerApp as Application>::restore(&mut cursor).expect("restore");
731
732 let mut reports_after = reports_before.clone();
736 reports_after.clear();
737 after.0.execute(
738 Symbol(1),
739 Order {
740 id: OrderId(2),
741 account: AccountId(1),
742 side: Side::Buy,
743 order_type: OrderType::Limit {
744 price: price(99),
745 post_only: false,
746 },
747 quantity: qty(5),
748 time_in_force: TimeInForce::GTC,
749 stp: SelfTradeProtection::Allow,
750 expiry_ns: 0,
751 },
752 &mut reports_after,
753 );
754 assert!(
755 !reports_after.is_empty(),
756 "restored exchange must accept orders"
757 );
758 }
759}