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