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

nautilus_kraken/websocket/dispatch/
mod.rs
