nautilus_kraken/websocket/dispatch/mod.rs
1// -------------------------------------------------------------------------------------------------
2// Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3// https://nautechsystems.io
4//
5// Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6// You may not use this file except in compliance with the License.
7// You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! WebSocket execution dispatch for the Kraken Spot and Futures clients.
17//!
18//! Implements the two-tier execution dispatch contract from
19//! `docs/developer_guide/adapters.md#tracked-and-external-execution-updates`:
20//!
21//! 1. The execution client registers an [`OrderIdentity`] in [`WsDispatchState`]
22//! when it submits an order.
23//! 2. WebSocket execution messages are routed through the per-product dispatch
24//! functions in [`futures`] and [`spot`]. For tracked orders the dispatch
25//! builds typed [`OrderEventAny`] events and emits them directly via
26//! [`ExecutionEventEmitter::send_order_event`]. For untracked / external
27//! orders the dispatch falls back to `OrderStatusReport` / `FillReport`
28//! so the engine can reconcile.
29//!
30//! The dispatch state lives in an `Arc<WsDispatchState>` shared between the
31//! main client thread (which registers identities at submission time) and the
32//! spawned WebSocket consumer task. `DashMap`/`DashSet` provide lock-free
33//! concurrent access.
34
35pub mod futures;
36pub mod spot;
37pub mod spot_orders;
38
39use std::sync::{
40 Arc,
41 atomic::{AtomicBool, Ordering},
42};
43
44use dashmap::{DashMap, DashSet};
45use indexmap::IndexSet;
46use nautilus_core::{AtomicMap, UUID4, UnixNanos};
47use nautilus_live::ExecutionEventEmitter;
48use nautilus_model::{
49 enums::{OrderSide, OrderType},
50 events::{OrderAccepted, OrderEventAny, OrderFilled},
51 identifiers::{
52 AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
53 },
54 reports::FillReport,
55 types::{Currency, Quantity},
56};
57use parking_lot::Mutex;
58use rust_decimal::Decimal;
59
60const DEDUP_CAPACITY: usize = 10_000;
61
62/// A terminal open-orders removal whose fills have not all arrived yet.
63///
64/// An `is_cancel=true` delta with reason `partial_fill` removes an order from
65/// the book with its remainder discarded (a converted Maker Protection hold
66/// or an IOC-style order). The venue's cumulative filled at the removal is
67/// parked here until the fills feed accounts it, so the closing cancel can
68/// never overtake the fill it belongs with.
69#[derive(Debug, Clone)]
70pub struct PendingRemoval {
71 /// Cumulative filled quantity reported by the venue on the removal delta.
72 pub venue_filled: Quantity,
73 /// Venue reason carried by the removal delta.
74 pub reason: Option<String>,
75 /// Venue event time of the removal delta.
76 pub ts_event: UnixNanos,
77}
78
79/// Snapshot of the mutable fields seen on a tracked `OpenOrdersDelta`.
80///
81/// Used by the futures delta path to discriminate partial fills (filled
82/// increased), modify acknowledgements (qty / price / trigger_price changed),
83/// and no-op deltas (nothing changed) when a follow-up delta arrives.
84#[derive(Debug, Clone, Copy, PartialEq)]
85pub struct DeltaSnapshot {
86 pub qty: Quantity,
87 pub filled: Quantity,
88 pub limit_price: Option<Decimal>,
89 pub stop_price: Option<Decimal>,
90}
91
92impl DeltaSnapshot {
93 pub(crate) fn new(
94 qty: Quantity,
95 filled: Quantity,
96 limit_price: Option<Decimal>,
97 stop_price: Option<Decimal>,
98 ) -> Self {
99 Self {
100 qty,
101 filled,
102 limit_price,
103 stop_price,
104 }
105 }
106
107 pub(crate) fn non_fill_fields_match(&self, other: &Self) -> bool {
108 self.qty == other.qty
109 && self.limit_price == other.limit_price
110 && self.stop_price == other.stop_price
111 }
112}
113
114/// Identity metadata captured when an order is submitted through this client.
115///
116/// Stored in [`WsDispatchState::order_identities`] keyed by the full Nautilus
117/// [`ClientOrderId`]. The dispatch functions use the identity to build typed
118/// order events for tracked orders without needing access to the engine cache
119/// (which is `!Send` and unreachable from the spawned WebSocket task).
120#[derive(Debug, Clone)]
121pub struct OrderIdentity {
122 /// Strategy that owns the order.
123 pub strategy_id: StrategyId,
124 /// Instrument the order targets.
125 pub instrument_id: InstrumentId,
126 /// Order side captured at submission.
127 pub order_side: OrderSide,
128 /// Order type captured at submission.
129 pub order_type: OrderType,
130 /// Order quantity captured at submission. Used to detect terminal fills.
131 pub quantity: Quantity,
132}
133
134/// Per-client dispatch state shared between order submission and the
135/// WebSocket consumer task.
136///
137/// Tracks which orders were submitted through this client (so we can route
138/// venue events to typed [`OrderEventAny`] emissions for tracked orders, and
139/// fall back to reports for external orders), and provides cross-stream
140/// dedup for `OrderAccepted` and `OrderFilled` emissions.
141#[derive(Debug)]
142pub struct WsDispatchState {
143 /// Tracked orders keyed by full Nautilus [`ClientOrderId`].
144 pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
145 /// Client order IDs for which an `OrderAccepted` event has been emitted.
146 pub emitted_accepted: DashSet<ClientOrderId>,
147 /// Client order IDs that have reached the filled terminal state.
148 pub filled_orders: DashSet<ClientOrderId>,
149 /// Symbol captured from execution frames for a venue order id.
150 ///
151 /// Kraken's spot v2 executions channel sends a `pending_new` frame with
152 /// full order details, then follow-up frames (`new`, `amended`,
153 /// `restated`, `status`) that omit fields which have not changed -
154 /// Kraken's docs show `symbol` omitted on the `new` delta. The dispatch
155 /// needs the symbol to resolve the instrument, so we cache it here from
156 /// any frame that carries it (first writer wins). Keyed by venue
157 /// `order_id` because delta frames often lack `cl_ord_id` as well.
158 ///
159 /// Known limitation: the live spot execution client currently subscribes
160 /// with `snap_orders=false` (see `execution/spot.rs`). Orders that were
161 /// already open at the venue before this process connected therefore do
162 /// not receive an in-session `pending_new`, and if their next delta
163 /// frame omits `symbol` it is dropped at symbol resolution. State for
164 /// such orders is recovered via REST reconciliation
165 /// (`request_order_status_reports`). Enabling `snap_orders=true` would
166 /// allow the executions snapshot to seed the cache for pre-existing
167 /// orders.
168 ///
169 /// Steady-state eviction happens on terminal exec types
170 /// (`Canceled`/`Filled`/`Expired`). Bounded by `DEDUP_CAPACITY` as a
171 /// safety net so missed terminal frames (reconnects, partial replays)
172 /// cannot leak entries indefinitely.
173 pub order_symbol_cache: DashMap<String, String>,
174 /// `ClientOrderId` captured from execution frames for a venue order id.
175 ///
176 /// Kraken's `pending_new` echoes our submitted `cl_ord_id`, but follow-up
177 /// delta frames (`new`, `amended`, `restated`, `status`) routinely omit
178 /// it. Without this mapping the dispatch cannot resolve the tracked
179 /// order from a delta and falls back to the untracked report path,
180 /// which loses the typed `OrderAccepted` event - the symptom behind
181 /// issue #4051.
182 ///
183 /// Populated whenever a frame resolves a `cl_ord_id` (first writer
184 /// wins, keyed by venue `order_id`). Consulted when `exec.cl_ord_id`
185 /// is `None`. Mirrors the `venue_client_map` used by the futures
186 /// dispatch path.
187 ///
188 /// Eviction policy matches `order_symbol_cache`: cleared on terminal
189 /// exec types and bounded by `DEDUP_CAPACITY` as a safety net.
190 pub order_client_id_cache: DashMap<String, ClientOrderId>,
191 /// Last snapshot of qty / filled / price / trigger_price seen on a
192 /// tracked `OpenOrdersDelta`.
193 ///
194 /// The futures delta path uses this map to discriminate partial-fill
195 /// notifications (the new delta carries `filled` greater than the
196 /// previously seen value), modify acknowledgements (a non-fill field
197 /// changed), and pure no-op deltas (nothing changed). It is updated only
198 /// by the delta path so that the fill path's own cumulative is not
199 /// double-counted.
200 pub delta_snapshots: DashMap<ClientOrderId, DeltaSnapshot>,
201 /// Cumulative filled quantity per tracked client order id, populated by
202 /// the fill side of dispatch.
203 ///
204 /// Compared against `OrderIdentity::quantity` to decide when to clean up
205 /// tracked state on a terminal fill.
206 pub order_filled_qty: DashMap<ClientOrderId, Quantity>,
207 /// Terminal part-fill removals waiting for their fills to be accounted,
208 /// populated by the futures delta side of dispatch and resolved by the
209 /// fill side.
210 pub pending_removals: DashMap<ClientOrderId, PendingRemoval>,
211 /// Trade IDs for which an `OrderFilled` event has been emitted.
212 ///
213 /// Bounded FIFO dedup: when capacity is reached, the oldest entry is
214 /// evicted on the next insert. A simple `clear()` at the threshold would
215 /// drop all recent trade IDs at once, opening a window where a reconnect
216 /// or replay immediately after the rollover could re-emit duplicate
217 /// `OrderFilled` events.
218 pub emitted_trades: Mutex<IndexSet<TradeId>>,
219 clearing: AtomicBool,
220}
221
222impl Default for WsDispatchState {
223 fn default() -> Self {
224 Self {
225 order_identities: DashMap::new(),
226 emitted_accepted: DashSet::default(),
227 filled_orders: DashSet::default(),
228 order_symbol_cache: DashMap::new(),
229 order_client_id_cache: DashMap::new(),
230 delta_snapshots: DashMap::new(),
231 order_filled_qty: DashMap::new(),
232 pending_removals: DashMap::new(),
233 emitted_trades: Mutex::new(IndexSet::with_capacity(DEDUP_CAPACITY)),
234 clearing: AtomicBool::new(false),
235 }
236 }
237}
238
239impl WsDispatchState {
240 /// Creates a new empty dispatch state.
241 #[must_use]
242 pub fn new() -> Self {
243 Self::default()
244 }
245
246 /// Registers an order identity. Called by the execution client at order
247 /// submission time, before any WebSocket events for the order can arrive.
248 pub fn register_identity(&self, client_order_id: ClientOrderId, identity: OrderIdentity) {
249 self.order_identities.insert(client_order_id, identity);
250 }
251
252 /// Returns a clone of the identity for the given client order id, if any.
253 #[must_use]
254 pub fn lookup_identity(&self, client_order_id: &ClientOrderId) -> Option<OrderIdentity> {
255 self.order_identities
256 .get(client_order_id)
257 .map(|r| r.clone())
258 }
259
260 /// Atomically marks an `OrderAccepted` event as emitted for this order.
261 ///
262 /// Returns `true` when the entry was newly inserted (caller should emit the
263 /// event), and `false` when an entry was already present (caller should
264 /// skip emission). Replaces the racier "contains-then-insert" pattern,
265 /// which allowed concurrent emitters to both observe `false` from
266 /// `contains` and then both insert + emit duplicate `OrderAccepted` events.
267 pub fn insert_accepted(&self, cid: ClientOrderId) -> bool {
268 self.evict_if_full(&self.emitted_accepted);
269 self.emitted_accepted.insert(cid)
270 }
271
272 /// Marks an order as having reached the filled terminal state.
273 pub fn insert_filled(&self, cid: ClientOrderId) {
274 self.evict_if_full(&self.filled_orders);
275 self.filled_orders.insert(cid);
276 }
277
278 /// Caches the symbol for a venue `order_id` if not already present.
279 ///
280 /// Atomic via [`DashMap::entry`] so concurrent callers cannot overwrite an
281 /// existing cached value - first writer wins, all later writers no-op.
282 /// The cheap `contains_key` fast path skips the key allocation when the
283 /// entry already exists; the `or_insert_with` covers the race that opens
284 /// between that check and the insert.
285 pub fn cache_order_symbol(&self, order_id: &str, symbol: &str) {
286 if self.order_symbol_cache.contains_key(order_id) {
287 return;
288 }
289 self.evict_map_if_full(&self.order_symbol_cache);
290 self.order_symbol_cache
291 .entry(order_id.to_string())
292 .or_insert_with(|| symbol.to_string());
293 }
294
295 /// Returns the symbol previously cached for a venue `order_id`, if any.
296 #[must_use]
297 pub fn lookup_order_symbol(&self, order_id: &str) -> Option<String> {
298 self.order_symbol_cache
299 .get(order_id)
300 .map(|r| r.value().clone())
301 }
302
303 /// Removes any cached symbol for a venue `order_id`. Called when the order
304 /// reaches a terminal state on the executions stream.
305 pub fn forget_order_symbol(&self, order_id: &str) {
306 self.order_symbol_cache.remove(order_id);
307 }
308
309 /// Caches the resolved `ClientOrderId` for a venue `order_id` if not
310 /// already present.
311 ///
312 /// Atomic via [`DashMap::entry`] so concurrent callers cannot overwrite an
313 /// existing cached value - first writer wins. The cheap `contains_key`
314 /// fast path skips the key allocation when the entry already exists.
315 pub fn cache_order_client_id(&self, order_id: &str, client_order_id: ClientOrderId) {
316 if self.order_client_id_cache.contains_key(order_id) {
317 return;
318 }
319 self.evict_map_if_full(&self.order_client_id_cache);
320 self.order_client_id_cache
321 .entry(order_id.to_string())
322 .or_insert(client_order_id);
323 }
324
325 /// Returns the `ClientOrderId` previously cached for a venue `order_id`,
326 /// if any.
327 #[must_use]
328 pub fn lookup_order_client_id(&self, order_id: &str) -> Option<ClientOrderId> {
329 self.order_client_id_cache.get(order_id).map(|r| *r)
330 }
331
332 /// Removes any cached `ClientOrderId` for a venue `order_id`. Called when
333 /// the order reaches a terminal state on the executions stream.
334 pub fn forget_order_client_id(&self, order_id: &str) {
335 self.order_client_id_cache.remove(order_id);
336 }
337
338 /// Atomically inserts a trade id into the dedup set.
339 ///
340 /// Returns `true` when the trade was already present (i.e. it is a
341 /// duplicate), `false` otherwise. When the dedup set is at capacity the
342 /// oldest entry is evicted to make room, preserving the `DEDUP_CAPACITY`
343 /// most recently seen trade IDs.
344 pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
345 let mut set = self.emitted_trades.lock();
346
347 if set.contains(&trade_id) {
348 return true;
349 }
350
351 if set.len() >= DEDUP_CAPACITY {
352 set.shift_remove_index(0);
353 }
354
355 set.insert(trade_id);
356 false
357 }
358
359 /// Removes all dispatch state for an order that has reached a terminal state.
360 pub fn cleanup_terminal(&self, client_order_id: &ClientOrderId) {
361 self.order_identities.remove(client_order_id);
362 self.emitted_accepted.remove(client_order_id);
363 self.order_filled_qty.remove(client_order_id);
364 self.delta_snapshots.remove(client_order_id);
365 self.pending_removals.remove(client_order_id);
366 }
367
368 /// Records cumulative filled quantity for a tracked order. Used by the
369 /// fill side of dispatch only.
370 pub fn record_filled_qty(&self, client_order_id: ClientOrderId, qty: Quantity) {
371 self.order_filled_qty.insert(client_order_id, qty);
372 }
373
374 /// Returns the previously recorded cumulative filled quantity, if any.
375 #[must_use]
376 pub fn previous_filled_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
377 self.order_filled_qty.get(client_order_id).map(|r| *r)
378 }
379
380 /// Parks a terminal part-fill removal until the fills feed accounts it.
381 pub fn insert_pending_removal(&self, client_order_id: ClientOrderId, removal: PendingRemoval) {
382 self.evict_map_if_full(&self.pending_removals);
383 self.pending_removals.insert(client_order_id, removal);
384 }
385
386 /// Returns a clone of the parked removal for the order, if any.
387 #[must_use]
388 pub fn pending_removal(&self, client_order_id: &ClientOrderId) -> Option<PendingRemoval> {
389 self.pending_removals
390 .get(client_order_id)
391 .map(|r| r.clone())
392 }
393
394 /// Removes the parked removal for the order, if any.
395 pub fn remove_pending_removal(&self, client_order_id: &ClientOrderId) {
396 self.pending_removals.remove(client_order_id);
397 }
398
399 /// Records the latest delta snapshot for a tracked order. Used by the
400 /// delta side of dispatch only.
401 pub fn record_delta_snapshot(&self, client_order_id: ClientOrderId, snapshot: DeltaSnapshot) {
402 self.delta_snapshots.insert(client_order_id, snapshot);
403 }
404
405 /// Returns the previously recorded delta snapshot, if any.
406 #[must_use]
407 pub fn previous_delta_snapshot(
408 &self,
409 client_order_id: &ClientOrderId,
410 ) -> Option<DeltaSnapshot> {
411 self.delta_snapshots.get(client_order_id).map(|r| *r)
412 }
413
414 /// Updates the tracked `quantity` for an order following a successful
415 /// modify acknowledgement, leaving all other identity fields untouched.
416 pub fn update_identity_quantity(&self, client_order_id: &ClientOrderId, quantity: Quantity) {
417 if let Some(mut entry) = self.order_identities.get_mut(client_order_id) {
418 entry.quantity = quantity;
419 }
420 }
421
422 fn evict_if_full(&self, set: &DashSet<ClientOrderId>) {
423 if set.len() >= DEDUP_CAPACITY
424 && self
425 .clearing
426 .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
427 .is_ok()
428 {
429 set.clear();
430 self.clearing.store(false, Ordering::Release);
431 }
432 }
433
434 fn evict_map_if_full<K, V>(&self, map: &DashMap<K, V>)
435 where
436 K: Eq + std::hash::Hash,
437 {
438 if map.len() >= DEDUP_CAPACITY
439 && self
440 .clearing
441 .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
442 .is_ok()
443 {
444 map.clear();
445 self.clearing.store(false, Ordering::Release);
446 }
447 }
448}
449
450/// Resolves a Kraken-truncated client order id to its full Nautilus form.
451///
452/// Kraken truncates non-UUID client order ids to 18 characters; the truncation
453/// map is populated at submission time so the WebSocket consumer can recover
454/// the original id. Falls back to constructing a fresh `ClientOrderId` from
455/// the truncated string when no mapping exists (the order is then treated as
456/// external by downstream lookup).
457pub(crate) fn resolve_client_order_id(
458 truncated: &str,
459 truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
460) -> ClientOrderId {
461 truncated_id_map
462 .load()
463 .get(truncated)
464 .copied()
465 .unwrap_or_else(|| ClientOrderId::new(truncated))
466}
467
468/// Synthesizes and emits an `OrderAccepted` event when one has not yet been
469/// emitted for the given order.
470///
471/// Used before emitting non-Accepted events (Filled, Canceled, Expired,
472/// Updated) so that strategies always observe the canonical
473/// `Submitted -> Accepted -> ...` lifecycle even when the venue compresses
474/// the acceptance and follow-up event into a single message (fast fills).
475#[expect(clippy::too_many_arguments)]
476pub(crate) fn ensure_accepted_emitted(
477 client_order_id: ClientOrderId,
478 venue_order_id: VenueOrderId,
479 account_id: AccountId,
480 identity: &OrderIdentity,
481 state: &WsDispatchState,
482 emitter: &ExecutionEventEmitter,
483 ts_event: UnixNanos,
484 ts_init: UnixNanos,
485) {
486 if !state.insert_accepted(client_order_id) {
487 return;
488 }
489 let accepted = OrderAccepted::new(
490 emitter.trader_id(),
491 identity.strategy_id,
492 identity.instrument_id,
493 client_order_id,
494 venue_order_id,
495 account_id,
496 UUID4::new(),
497 ts_event,
498 ts_init,
499 false,
500 );
501 emitter.send_order_event(OrderEventAny::Accepted(accepted));
502}
503
504/// Builds an [`OrderFilled`] event from a [`FillReport`] and tracked
505/// [`OrderIdentity`].
506pub(crate) fn fill_report_to_order_filled(
507 report: &FillReport,
508 trader_id: TraderId,
509 identity: &OrderIdentity,
510 quote_currency: Currency,
511 client_order_id: ClientOrderId,
512) -> OrderFilled {
513 OrderFilled::new(
514 trader_id,
515 identity.strategy_id,
516 identity.instrument_id,
517 client_order_id,
518 report.venue_order_id,
519 report.account_id,
520 report.trade_id,
521 identity.order_side,
522 identity.order_type,
523 report.last_qty,
524 report.last_px,
525 quote_currency,
526 report.liquidity_side,
527 UUID4::new(),
528 report.ts_event,
529 report.ts_init,
530 false,
531 report.venue_position_id,
532 Some(report.commission),
533 None,
534 )
535}
536
537#[cfg(test)]
538mod tests {
539 use nautilus_model::{
540 enums::{OrderSide, OrderType},
541 identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId},
542 };
543 use rstest::rstest;
544
545 use super::*;
546
547 fn make_identity() -> OrderIdentity {
548 OrderIdentity {
549 strategy_id: StrategyId::new("EXEC_TESTER-001"),
550 instrument_id: InstrumentId::from("PF_XBTUSD.KRAKEN"),
551 order_side: OrderSide::Buy,
552 order_type: OrderType::Limit,
553 quantity: Quantity::from("0.0001"),
554 }
555 }
556
557 #[rstest]
558 fn test_register_and_lookup_identity() {
559 let state = WsDispatchState::new();
560 let cid = ClientOrderId::new("uuid-1");
561 state.register_identity(cid, make_identity());
562
563 let found = state.lookup_identity(&cid);
564 assert!(found.is_some());
565 let identity = found.unwrap();
566 assert_eq!(identity.strategy_id.as_str(), "EXEC_TESTER-001");
567 assert_eq!(identity.order_side, OrderSide::Buy);
568 }
569
570 #[rstest]
571 fn test_lookup_identity_missing_returns_none() {
572 let state = WsDispatchState::new();
573 let cid = ClientOrderId::new("not-tracked");
574 assert!(state.lookup_identity(&cid).is_none());
575 }
576
577 #[rstest]
578 fn test_insert_accepted_dedup() {
579 let state = WsDispatchState::new();
580 let cid = ClientOrderId::new("uuid-2");
581 assert!(!state.emitted_accepted.contains(&cid));
582 state.insert_accepted(cid);
583 assert!(state.emitted_accepted.contains(&cid));
584 // Second insert is a no-op.
585 state.insert_accepted(cid);
586 assert!(state.emitted_accepted.contains(&cid));
587 }
588
589 #[rstest]
590 fn test_insert_accepted_returns_true_on_first_insert_false_on_repeat() {
591 let state = WsDispatchState::new();
592 let cid = ClientOrderId::new("uuid-atomic");
593
594 assert!(
595 state.insert_accepted(cid),
596 "first insert must report newly inserted",
597 );
598 assert!(
599 !state.insert_accepted(cid),
600 "second insert must report already present (atomic dedup)",
601 );
602 }
603
604 #[rstest]
605 fn test_cache_order_symbol_first_write_wins() {
606 let state = WsDispatchState::new();
607 state.cache_order_symbol("v-order-1", "BTC/USD");
608 state.cache_order_symbol("v-order-1", "ETH/USD");
609
610 assert_eq!(
611 state.lookup_order_symbol("v-order-1").as_deref(),
612 Some("BTC/USD"),
613 "later writes must not overwrite the original cached symbol",
614 );
615 }
616
617 #[rstest]
618 fn test_cache_order_symbol_bounded_by_capacity() {
619 let state = WsDispatchState::new();
620 for i in 0..DEDUP_CAPACITY {
621 state.cache_order_symbol(format!("v-order-{i}").as_str(), "BTC/USD");
622 }
623 // At capacity; the next insert triggers a `clear()` before insertion.
624 state.cache_order_symbol("v-order-overflow", "BTC/USD");
625
626 assert!(
627 state.order_symbol_cache.len() <= DEDUP_CAPACITY,
628 "cache must stay within DEDUP_CAPACITY after overflow",
629 );
630 assert_eq!(
631 state.lookup_order_symbol("v-order-overflow").as_deref(),
632 Some("BTC/USD"),
633 "the overflow entry was inserted after eviction",
634 );
635 }
636
637 #[rstest]
638 fn test_check_and_insert_trade_detects_duplicates() {
639 let state = WsDispatchState::new();
640 let trade = TradeId::new("trade-1");
641 // First insert: not a duplicate.
642 assert!(!state.check_and_insert_trade(trade));
643 // Second insert: duplicate.
644 assert!(state.check_and_insert_trade(trade));
645 }
646
647 #[rstest]
648 fn test_check_and_insert_trade_fifo_eviction_preserves_recent_ids() {
649 // Verifies the dedup window slides rather than collapsing to zero at
650 // the capacity threshold. Overshooting by one entry must evict only
651 // the oldest (`trade-0`), leaving every other ID still deduped.
652 let state = WsDispatchState::new();
653 for i in 0..DEDUP_CAPACITY {
654 let trade = TradeId::new(format!("trade-{i}").as_str());
655 assert!(!state.check_and_insert_trade(trade));
656 }
657 // At capacity; the next insert evicts `trade-0`.
658 let overflow = TradeId::new(format!("trade-{DEDUP_CAPACITY}").as_str());
659 assert!(!state.check_and_insert_trade(overflow));
660
661 // Inspect the dedup set directly to confirm FIFO behavior without
662 // perturbing state via another `check_and_insert_trade` call.
663 let set = state.emitted_trades.lock();
664 assert_eq!(set.len(), DEDUP_CAPACITY);
665 assert!(
666 !set.contains(&TradeId::new("trade-0")),
667 "oldest entry should have been evicted",
668 );
669 assert!(
670 set.contains(&TradeId::new("trade-1")),
671 "second-oldest remains"
672 );
673 assert!(
674 set.contains(&TradeId::new(
675 format!("trade-{}", DEDUP_CAPACITY - 1).as_str(),
676 )),
677 "most-recent pre-overflow entry remains",
678 );
679 assert!(
680 set.contains(&overflow),
681 "the overflow entry was inserted after eviction",
682 );
683 }
684
685 #[rstest]
686 fn test_cleanup_terminal_removes_state() {
687 let state = WsDispatchState::new();
688 let cid = ClientOrderId::new("uuid-3");
689 state.register_identity(cid, make_identity());
690 state.insert_accepted(cid);
691 state.record_filled_qty(cid, Quantity::from("0.0001"));
692 state.insert_pending_removal(
693 cid,
694 PendingRemoval {
695 venue_filled: Quantity::from("0.0001"),
696 reason: Some("partial_fill".to_string()),
697 ts_event: UnixNanos::default(),
698 },
699 );
700
701 assert!(state.lookup_identity(&cid).is_some());
702 assert!(state.emitted_accepted.contains(&cid));
703 assert!(state.pending_removal(&cid).is_some());
704
705 state.cleanup_terminal(&cid);
706
707 assert!(state.lookup_identity(&cid).is_none());
708 assert!(!state.emitted_accepted.contains(&cid));
709 assert!(state.previous_filled_qty(&cid).is_none());
710 assert!(state.pending_removal(&cid).is_none());
711 }
712
713 #[rstest]
714 fn test_resolve_client_order_id_via_truncated_map() {
715 let map: Arc<AtomicMap<String, ClientOrderId>> = Arc::new(AtomicMap::new());
716 let full = ClientOrderId::new("full-uuid-12345");
717 map.insert("trunc-id".to_string(), full);
718
719 let resolved = resolve_client_order_id("trunc-id", &map);
720 assert_eq!(resolved, full);
721 }
722
723 #[rstest]
724 fn test_resolve_client_order_id_falls_back_to_input() {
725 let map: Arc<AtomicMap<String, ClientOrderId>> = Arc::new(AtomicMap::new());
726 let resolved = resolve_client_order_id("unknown", &map);
727 assert_eq!(resolved.as_str(), "unknown");
728 }
729}