nautilus_derive/websocket/
dispatch.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//! Shared state for the Derive execution WebSocket dispatch loop.
17//!
18//! Holds instrument precision, identity context for submitted orders and active
19//! orders restored from the cache, and the cross-stream deduplication gates that
20//! keep replay frames and concurrent `.orders` / `.trades` updates from emitting
21//! duplicate events.
22//!
23//! Tracked orders (those whose identity was registered at submission or restored)
24//! produce proper order events (`OrderAccepted`, `OrderFilled`, `OrderCanceled`,
25//! `OrderExpired`, `OrderRejected`). Untracked frames fall back to execution
26//! reports for downstream reconciliation.
27
28use std::sync::Arc;
29
30use ahash::{AHashMap, AHashSet};
31use nautilus_common::cache::fifo::{FifoCache, FifoCacheMap};
32use nautilus_model::{
33    enums::{OrderSide, OrderStatus, OrderType},
34    events::OrderEventAny,
35    identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId, VenueOrderId},
36    orders::{Order, OrderAny},
37    types::{Price, Quantity},
38};
39use parking_lot::{Mutex, ReentrantMutex, ReentrantMutexGuard};
40use rust_decimal::Decimal;
41
42use crate::http::models::{DeriveReplaceResult, DeriveTrade};
43
44/// Capacity for the cross-source trade-id dedup cache. Sized to cover any
45/// reconciliation lookback window plausible for live trading.
46pub const TRADE_DEDUP_CAPACITY: usize = 4_096;
47
48/// Capacity for the per-order accepted / filled dedup caches. Tracks active
49/// and recently-terminal orders so reconnect replays do not re-emit lifecycle
50/// events; need only span the live-stream replay window plus a margin.
51pub const ORDER_DEDUP_CAPACITY: usize = 1_024;
52
53/// Order identity captured at submission or restored so the dispatch task can build
54/// proper order events without consulting the cache.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct OrderIdentity {
57    pub instrument_id: InstrumentId,
58    pub strategy_id: StrategyId,
59    pub order_side: OrderSide,
60    pub order_type: OrderType,
61}
62
63#[derive(Debug, Clone)]
64pub(crate) struct ReplaceCompletion {
65    pub old_venue_order_id: VenueOrderId,
66    pub nonce: u64,
67    pub identity: OrderIdentity,
68    pub result: Result<DeriveReplaceResult, String>,
69    pub trades: Vec<DeriveTrade>,
70}
71
72/// Shared dispatch state for the Derive WS execution loop.
73///
74/// Instrument precision populates from instrument definitions and governs
75/// execution report values. `order_identities` populates on submission and from
76/// active cached orders before streaming, and is consulted by both the `.orders`
77/// and `.trades` dispatch paths to decide whether a frame belongs to a tracked or external
78/// order. `pending_modifies` and `bound_venue_order_ids` track the in-flight
79/// and current venue order id of a `private/replace` so the dispatch suppresses
80/// events for the superseded leg.
81#[derive(Debug)]
82pub struct WsDispatchState {
83    subaccount_id: u64,
84    delivery: ReentrantMutex<()>,
85    order_identities: Mutex<AHashMap<ClientOrderId, OrderIdentity>>,
86    instrument_precisions: Mutex<AHashMap<InstrumentId, (u8, u8)>>,
87    emitted_accepted: Mutex<AHashSet<ClientOrderId>>,
88    triggers_active: Mutex<AHashSet<ClientOrderId>>,
89    terminal_orders: Mutex<FifoCacheMap<ClientOrderId, OrderBinding, ORDER_DEDUP_CAPACITY>>,
90    terminal_revision: Mutex<Arc<()>>,
91    #[expect(
92        clippy::type_complexity,
93        reason = "order quantity, cumulative fills, and price share one lock for atomic updates"
94    )]
95    order_shapes: Mutex<AHashMap<ClientOrderId, (Quantity, Decimal, Option<Price>)>>,
96    emitted_canceled: Mutex<FifoCache<ClientOrderId, ORDER_DEDUP_CAPACITY>>,
97    filled_orders: Mutex<FifoCache<ClientOrderId, ORDER_DEDUP_CAPACITY>>,
98    emitted_trades: Mutex<FifoCache<TradeId, TRADE_DEDUP_CAPACITY>>,
99    bound_venue_order_ids: Mutex<AHashMap<ClientOrderId, VenueOrderId>>,
100    pending_modifies: Mutex<AHashMap<ClientOrderId, VenueOrderId>>,
101    modify_nonces: Mutex<AHashMap<ClientOrderId, u64>>,
102    deferred_trades: Mutex<AHashMap<ClientOrderId, Vec<DeriveTrade>>>,
103    undelivered_trades: Mutex<AHashMap<(String, String), DeriveTrade>>,
104    replace_completions: Mutex<AHashMap<ClientOrderId, ReplaceCompletion>>,
105    venue_order_legs: Mutex<AHashMap<ClientOrderId, AHashSet<VenueOrderId>>>,
106    modify_targets: Mutex<AHashMap<ClientOrderId, (Quantity, Option<Price>)>>,
107    unresolved_bindings: Mutex<AHashMap<ClientOrderId, AHashSet<VenueOrderId>>>,
108}
109
110impl WsDispatchState {
111    /// Creates dispatch state scoped to one native Derive subaccount.
112    #[must_use]
113    pub fn new(subaccount_id: u64) -> Self {
114        Self {
115            subaccount_id,
116            delivery: Default::default(),
117            order_identities: Default::default(),
118            instrument_precisions: Default::default(),
119            emitted_accepted: Default::default(),
120            triggers_active: Default::default(),
121            terminal_orders: Default::default(),
122            terminal_revision: Default::default(),
123            order_shapes: Default::default(),
124            emitted_canceled: Default::default(),
125            filled_orders: Default::default(),
126            emitted_trades: Default::default(),
127            bound_venue_order_ids: Default::default(),
128            pending_modifies: Default::default(),
129            modify_nonces: Default::default(),
130            deferred_trades: Default::default(),
131            undelivered_trades: Default::default(),
132            replace_completions: Default::default(),
133            venue_order_legs: Default::default(),
134            modify_targets: Default::default(),
135            unresolved_bindings: Default::default(),
136        }
137    }
138
139    pub(crate) const fn subaccount_id(&self) -> u64 {
140        self.subaccount_id
141    }
142
143    pub(crate) fn owns_subaccount(&self, subaccount_id: i64) -> bool {
144        u64::try_from(subaccount_id).ok() == Some(self.subaccount_id)
145    }
146
147    pub(crate) fn delivery_guard(&self) -> ReentrantMutexGuard<'_, ()> {
148        self.delivery.lock()
149    }
150
151    /// Records price and size precision for execution report parsing.
152    pub(crate) fn register_instrument_precision(
153        &self,
154        instrument_id: InstrumentId,
155        price_precision: u8,
156        size_precision: u8,
157    ) {
158        self.instrument_precisions
159            .lock()
160            .insert(instrument_id, (price_precision, size_precision));
161    }
162
163    /// Returns price and size precision for an instrument, when registered.
164    #[must_use]
165    pub(crate) fn instrument_precision(&self, instrument_id: &InstrumentId) -> Option<(u8, u8)> {
166        self.instrument_precisions
167            .lock()
168            .get(instrument_id)
169            .copied()
170    }
171
172    /// Registers an order identity captured at submission so subsequent WS
173    /// frames for the same client_order_id resolve to the tracked path.
174    pub fn register_identity(&self, client_order_id: ClientOrderId, identity: OrderIdentity) {
175        self.terminal_orders.lock().remove(&client_order_id);
176        self.order_identities
177            .lock()
178            .insert(client_order_id, identity);
179    }
180
181    /// Returns the registered identity for a client order, when one was
182    /// captured at submission time.
183    #[must_use]
184    pub fn identity(&self, client_order_id: &ClientOrderId) -> Option<OrderIdentity> {
185        self.order_identities.lock().get(client_order_id).copied()
186    }
187
188    pub(crate) fn restore_order(&self, order: &OrderAny) {
189        let _delivery = self.delivery_guard();
190        let client_order_id = order.client_order_id();
191        if self.identity(&client_order_id).is_some()
192            || self.is_terminal(&client_order_id, order.instrument_id())
193        {
194            return;
195        }
196
197        self.register_identity(
198            client_order_id,
199            OrderIdentity {
200                instrument_id: order.instrument_id(),
201                strategy_id: order.strategy_id(),
202                order_side: order.order_side(),
203                order_type: order.order_type(),
204            },
205        );
206
207        self.order_shapes.lock().insert(
208            client_order_id,
209            (
210                order.filled_qty() + order.leaves_qty(),
211                order.filled_qty().as_decimal(),
212                order.price(),
213            ),
214        );
215
216        if let Some(venue_order_id) = order.venue_order_id() {
217            self.record_venue_order_id(client_order_id, venue_order_id);
218            if order.status() == OrderStatus::PendingUpdate {
219                self.mark_pending_modify(client_order_id, venue_order_id);
220            }
221        }
222
223        let mut current_filled = false;
224
225        for event in order.events() {
226            match event {
227                OrderEventAny::Accepted(accepted) => {
228                    self.mark_accepted(client_order_id);
229                    self.record_venue_leg(client_order_id, accepted.venue_order_id);
230                }
231                OrderEventAny::Updated(updated) => {
232                    if let Some(venue_order_id) = updated.venue_order_id {
233                        self.record_venue_leg(client_order_id, venue_order_id);
234                    }
235                }
236                OrderEventAny::Filled(fill) => {
237                    self.record_venue_leg(client_order_id, fill.venue_order_id);
238                    self.mark_accepted(client_order_id);
239                    self.check_and_insert_trade(fill.trade_id);
240                    current_filled |= order.venue_order_id() == Some(fill.venue_order_id);
241                }
242                _ => {}
243            }
244        }
245
246        if order.is_triggered() == Some(true) || (order.is_triggered().is_some() && current_filled)
247        {
248            self.mark_trigger_active(client_order_id);
249        }
250    }
251
252    /// Retires active ownership and preserves bounded terminal replay state.
253    pub fn forget(&self, client_order_id: &ClientOrderId) {
254        let _delivery = self.delivery_guard();
255
256        if let Some(binding) = self.order_binding(client_order_id) {
257            let mut terminal = self.terminal_orders.lock();
258            if terminal.len() == terminal.capacity() && !terminal.contains_key(client_order_id) {
259                *self.terminal_revision.lock() = Arc::new(());
260            }
261
262            terminal.insert(*client_order_id, binding);
263        }
264
265        self.order_identities.lock().remove(client_order_id);
266        self.order_shapes.lock().remove(client_order_id);
267        self.emitted_accepted.lock().remove(client_order_id);
268        self.triggers_active.lock().remove(client_order_id);
269        self.bound_venue_order_ids.lock().remove(client_order_id);
270        self.venue_order_legs.lock().remove(client_order_id);
271        self.pending_modifies.lock().remove(client_order_id);
272        self.modify_nonces.lock().remove(client_order_id);
273        self.modify_targets.lock().remove(client_order_id);
274        self.unresolved_bindings.lock().remove(client_order_id);
275    }
276
277    pub(crate) fn mark_trigger_active(&self, client_order_id: ClientOrderId) {
278        self.triggers_active.lock().insert(client_order_id);
279    }
280
281    pub(crate) fn trigger_active(&self, client_order_id: &ClientOrderId) -> bool {
282        let _delivery = self.delivery_guard();
283        self.triggers_active.lock().contains(client_order_id)
284    }
285
286    pub(crate) fn take_identity(&self, client_order_id: &ClientOrderId) -> Option<OrderIdentity> {
287        let identity = self.identity(client_order_id);
288        if identity.is_some() {
289            self.forget(client_order_id);
290        }
291
292        identity
293    }
294
295    pub(crate) fn terminal_revision(&self) -> Arc<()> {
296        Arc::clone(&self.terminal_revision.lock())
297    }
298
299    pub(crate) fn known_instrument(&self, client_order_id: &ClientOrderId) -> Option<InstrumentId> {
300        self.identity(client_order_id)
301            .map(|identity| identity.instrument_id)
302            .or_else(|| {
303                self.terminal_orders
304                    .lock()
305                    .get(client_order_id)
306                    .map(|binding| binding.identity.instrument_id)
307            })
308    }
309
310    pub(crate) fn is_terminal(
311        &self,
312        client_order_id: &ClientOrderId,
313        instrument_id: InstrumentId,
314    ) -> bool {
315        self.terminal_orders
316            .lock()
317            .get(client_order_id)
318            .is_some_and(|binding| binding.identity.instrument_id == instrument_id)
319    }
320
321    pub(crate) fn record_order_shape(
322        &self,
323        client_order_id: ClientOrderId,
324        quantity: Quantity,
325        price: Option<Price>,
326    ) {
327        self.order_shapes
328            .lock()
329            .entry(client_order_id)
330            .and_modify(|entry| {
331                entry.0 = quantity;
332                entry.2 = price;
333            })
334            .or_insert((quantity, Decimal::ZERO, price));
335    }
336
337    pub(crate) fn order_shape(
338        &self,
339        client_order_id: &ClientOrderId,
340    ) -> Option<(Quantity, Option<Price>)> {
341        self.order_shapes
342            .lock()
343            .get(client_order_id)
344            .map(|entry| (entry.0, entry.2))
345    }
346
347    pub(crate) fn record_fill(&self, client_order_id: ClientOrderId, quantity: Quantity) {
348        let _delivery = self.delivery_guard();
349
350        let complete = self
351            .order_shapes
352            .lock()
353            .get_mut(&client_order_id)
354            .is_some_and(|entry| {
355                entry.1 += quantity.as_decimal();
356                entry.1 >= entry.0.as_decimal()
357            });
358
359        if complete
360            && self.pending_modify(&client_order_id).is_none()
361            && !self.binding_unresolved(&client_order_id)
362        {
363            self.forget(&client_order_id);
364        }
365    }
366
367    /// Records the venue order id currently bound to a tracked client order.
368    pub fn record_venue_order_id(
369        &self,
370        client_order_id: ClientOrderId,
371        venue_order_id: VenueOrderId,
372    ) {
373        self.record_venue_leg(client_order_id, venue_order_id);
374        self.bound_venue_order_ids
375            .lock()
376            .insert(client_order_id, venue_order_id);
377    }
378
379    pub(crate) fn record_venue_leg(
380        &self,
381        client_order_id: ClientOrderId,
382        venue_order_id: VenueOrderId,
383    ) {
384        self.venue_order_legs
385            .lock()
386            .entry(client_order_id)
387            .or_default()
388            .insert(venue_order_id);
389    }
390
391    pub(crate) fn knows_venue_leg(
392        &self,
393        client_order_id: &ClientOrderId,
394        venue_order_id: VenueOrderId,
395    ) -> bool {
396        self.venue_order_legs
397            .lock()
398            .get(client_order_id)
399            .is_some_and(|legs| legs.contains(&venue_order_id))
400    }
401
402    pub(crate) fn venue_order_legs(&self, client_order_id: &ClientOrderId) -> Vec<VenueOrderId> {
403        self.venue_order_legs
404            .lock()
405            .get(client_order_id)
406            .map(|legs| legs.iter().copied().collect())
407            .unwrap_or_default()
408    }
409
410    pub(crate) fn order_binding(&self, client_order_id: &ClientOrderId) -> Option<OrderBinding> {
411        let _delivery = self.delivery_guard();
412        self.identity(client_order_id)
413            .map(|identity| {
414                let mut venue_order_legs = self.venue_order_legs(client_order_id);
415                venue_order_legs.sort_unstable();
416
417                OrderBinding {
418                    identity,
419                    venue_order_id: self.bound_venue_order_id(client_order_id),
420                    venue_order_legs,
421                }
422            })
423            .or_else(|| self.terminal_orders.lock().get(client_order_id).cloned())
424    }
425
426    pub(crate) fn defer_binding(
427        &self,
428        client_order_id: ClientOrderId,
429        venue_order_id: VenueOrderId,
430    ) {
431        self.unresolved_bindings
432            .lock()
433            .entry(client_order_id)
434            .or_default()
435            .insert(venue_order_id);
436    }
437
438    pub(crate) fn binding_unresolved(&self, client_order_id: &ClientOrderId) -> bool {
439        self.unresolved_bindings
440            .lock()
441            .get(client_order_id)
442            .is_some_and(|legs| {
443                legs.iter()
444                    .any(|venue_order_id| !self.knows_venue_leg(client_order_id, *venue_order_id))
445            })
446    }
447
448    pub(crate) fn unresolved_client_order_id(
449        &self,
450        instrument_id: Option<InstrumentId>,
451    ) -> Option<ClientOrderId> {
452        let mut ids: Vec<_> = self.unresolved_bindings.lock().keys().copied().collect();
453        ids.extend(self.pending_modifies.lock().keys().copied());
454        ids.sort_unstable();
455        ids.dedup();
456        ids.into_iter().find(|client_order_id| {
457            self.identity(client_order_id).is_some_and(|identity| {
458                instrument_id.is_none_or(|instrument_id| identity.instrument_id == instrument_id)
459            }) && (self.binding_unresolved(client_order_id)
460                || self.pending_modify(client_order_id).is_some())
461        })
462    }
463
464    pub(crate) fn has_active_orders(&self) -> bool {
465        !self.order_identities.lock().is_empty()
466    }
467
468    pub(crate) fn defer_trade(&self, client_order_id: ClientOrderId, trade: DeriveTrade) {
469        let mut deferred = self.deferred_trades.lock();
470        let trades = deferred.entry(client_order_id).or_default();
471        if !trades.iter().any(|existing| {
472            existing.trade_id == trade.trade_id && existing.order_id == trade.order_id
473        }) {
474            trades.push(trade);
475        }
476    }
477
478    pub(crate) fn take_deferred_trades(&self, client_order_id: &ClientOrderId) -> Vec<DeriveTrade> {
479        self.deferred_trades
480            .lock()
481            .remove(client_order_id)
482            .unwrap_or_default()
483    }
484
485    pub(crate) fn retain_trade(&self, trade: DeriveTrade) {
486        self.undelivered_trades
487            .lock()
488            .insert((trade.order_id.clone(), trade.trade_id.clone()), trade);
489    }
490
491    pub(crate) fn take_undelivered_trades(&self) -> Vec<DeriveTrade> {
492        let mut trades: Vec<_> = self
493            .undelivered_trades
494            .lock()
495            .drain()
496            .map(|(_, trade)| trade)
497            .collect();
498        trades.sort_unstable_by(|a, b| {
499            (a.timestamp, &a.order_id, &a.trade_id).cmp(&(b.timestamp, &b.order_id, &b.trade_id))
500        });
501
502        trades
503    }
504
505    pub(crate) fn record_replace_completion(
506        &self,
507        client_order_id: ClientOrderId,
508        completion: ReplaceCompletion,
509    ) {
510        self.replace_completions
511            .lock()
512            .insert(client_order_id, completion);
513    }
514
515    pub(crate) fn replace_completions(&self) -> Vec<(ClientOrderId, ReplaceCompletion)> {
516        let mut completions: Vec<_> = self
517            .replace_completions
518            .lock()
519            .iter()
520            .map(|(cid, completion)| (*cid, completion.clone()))
521            .collect();
522        completions.sort_unstable_by_key(|(cid, _)| *cid);
523        completions
524    }
525
526    pub(crate) fn clear_replace_completion(&self, client_order_id: &ClientOrderId) {
527        self.replace_completions.lock().remove(client_order_id);
528    }
529
530    /// Returns the venue order id currently bound to a tracked client order.
531    #[must_use]
532    pub fn bound_venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
533        self.bound_venue_order_ids
534            .lock()
535            .get(client_order_id)
536            .copied()
537    }
538
539    /// Records the old venue order id of an in-flight `private/replace`, set
540    /// before the request so the cancel leg is suppressed.
541    pub fn mark_pending_modify(
542        &self,
543        client_order_id: ClientOrderId,
544        old_venue_order_id: VenueOrderId,
545    ) -> bool {
546        let mut pending = self.pending_modifies.lock();
547        if pending.contains_key(&client_order_id) {
548            return false;
549        }
550
551        pending.insert(client_order_id, old_venue_order_id);
552        true
553    }
554
555    /// Clears the in-flight modify marker once the replace resolves.
556    pub fn clear_pending_modify(&self, client_order_id: &ClientOrderId) {
557        self.pending_modifies.lock().remove(client_order_id);
558        self.modify_nonces.lock().remove(client_order_id);
559        self.modify_targets.lock().remove(client_order_id);
560    }
561
562    pub(crate) fn record_modify_nonce(&self, client_order_id: ClientOrderId, nonce: u64) {
563        self.modify_nonces.lock().insert(client_order_id, nonce);
564    }
565
566    pub(crate) fn modify_nonce(&self, client_order_id: &ClientOrderId) -> Option<u64> {
567        self.modify_nonces.lock().get(client_order_id).copied()
568    }
569
570    pub(crate) fn record_modify_target(
571        &self,
572        client_order_id: ClientOrderId,
573        quantity: Quantity,
574        price: Option<Price>,
575    ) {
576        self.modify_targets
577            .lock()
578            .insert(client_order_id, (quantity, price));
579    }
580
581    pub(crate) fn modify_target(
582        &self,
583        client_order_id: &ClientOrderId,
584    ) -> Option<(Quantity, Option<Price>)> {
585        self.modify_targets.lock().get(client_order_id).copied()
586    }
587
588    pub(crate) fn take_modify_target(
589        &self,
590        client_order_id: &ClientOrderId,
591    ) -> Option<(Quantity, Option<Price>)> {
592        self.modify_targets.lock().remove(client_order_id)
593    }
594
595    /// Returns the old venue order id of an in-flight modify, when one is set.
596    #[must_use]
597    pub fn pending_modify(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
598        self.pending_modifies.lock().get(client_order_id).copied()
599    }
600
601    /// Rebinds an in-flight modify when a frame for its replacement arrives
602    /// before the `private/replace` response.
603    pub fn bind_incoming_modify(
604        &self,
605        client_order_id: ClientOrderId,
606        venue_order_id: VenueOrderId,
607        terminal: bool,
608    ) -> bool {
609        let mut pending = self.pending_modifies.lock();
610
611        let Some(old_venue_order_id) = pending.get(&client_order_id).copied() else {
612            return false;
613        };
614
615        if old_venue_order_id == venue_order_id {
616            return false;
617        }
618
619        let mut bound = self.bound_venue_order_ids.lock();
620        if bound
621            .get(&client_order_id)
622            .is_some_and(|current| *current != old_venue_order_id)
623        {
624            return false;
625        }
626
627        bound.insert(client_order_id, venue_order_id);
628        if terminal {
629            pending.remove(&client_order_id);
630        }
631
632        true
633    }
634
635    /// Atomically claims a pending modify for its RPC response, optionally
636    /// rebinding the replacement venue order id before clearing the marker.
637    /// Returns `false` when an incoming terminal frame already resolved it.
638    pub fn take_pending_modify(
639        &self,
640        client_order_id: &ClientOrderId,
641        old_venue_order_id: VenueOrderId,
642        new_venue_order_id: Option<VenueOrderId>,
643    ) -> bool {
644        let mut pending = self.pending_modifies.lock();
645        if pending.get(client_order_id) != Some(&old_venue_order_id) {
646            return false;
647        }
648
649        if let Some(new_venue_order_id) = new_venue_order_id {
650            self.record_venue_leg(*client_order_id, new_venue_order_id);
651            self.bound_venue_order_ids
652                .lock()
653                .insert(*client_order_id, new_venue_order_id);
654        }
655
656        pending.remove(client_order_id);
657        self.modify_nonces.lock().remove(client_order_id);
658        true
659    }
660
661    /// Returns `true` when an `OrderAccepted` has already been emitted for
662    /// this client order in the current process lifetime.
663    #[must_use]
664    pub fn contains_accepted(&self, client_order_id: &ClientOrderId) -> bool {
665        self.emitted_accepted.lock().contains(client_order_id)
666    }
667
668    /// Records that `OrderAccepted` has been emitted for this client order.
669    /// Returns `true` when the marker was already present (duplicate).
670    pub fn mark_accepted(&self, client_order_id: ClientOrderId) -> bool {
671        !self.emitted_accepted.lock().insert(client_order_id)
672    }
673
674    pub(crate) fn contains_canceled(&self, client_order_id: &ClientOrderId) -> bool {
675        self.emitted_canceled.lock().contains(client_order_id)
676    }
677
678    /// Records that `OrderCanceled` has been emitted for this client order.
679    /// Returns `true` when the marker was already present (duplicate).
680    pub fn mark_canceled(&self, client_order_id: ClientOrderId) -> bool {
681        let mut cache = self.emitted_canceled.lock();
682        if cache.contains(&client_order_id) {
683            return true;
684        }
685
686        cache.add(client_order_id);
687        false
688    }
689
690    /// Returns `true` when this client order has reached a terminal filled
691    /// state, used to suppress stale Accepted frames replayed on reconnect.
692    #[must_use]
693    pub fn contains_filled(&self, client_order_id: &ClientOrderId) -> bool {
694        self.filled_orders.lock().contains(client_order_id)
695    }
696
697    /// Marks the client order as terminally filled. Idempotent.
698    pub fn mark_filled(&self, client_order_id: ClientOrderId) {
699        let mut cache = self.filled_orders.lock();
700        if !cache.contains(&client_order_id) {
701            cache.add(client_order_id);
702        }
703    }
704
705    /// Inserts the trade id atomically. Returns `true` when the id was
706    /// already present (i.e., this fill should be skipped as a duplicate).
707    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
708        let mut cache = self.emitted_trades.lock();
709        if cache.contains(&trade_id) {
710            return true;
711        }
712
713        cache.add(trade_id);
714        false
715    }
716
717    /// Returns `true` when this trade id has already been seen, without
718    /// mutating state.
719    #[must_use]
720    pub fn contains_trade(&self, trade_id: &TradeId) -> bool {
721        self.emitted_trades.lock().contains(trade_id)
722    }
723}
724
725#[derive(Debug, Clone, PartialEq, Eq)]
726pub(crate) struct OrderBinding {
727    pub identity: OrderIdentity,
728    pub venue_order_id: Option<VenueOrderId>,
729    pub venue_order_legs: Vec<VenueOrderId>,
730}
731
732#[cfg(test)]
733mod tests {
734    use nautilus_model::{
735        enums::{OrderSide, OrderType},
736        identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId, VenueOrderId},
737    };
738    use rstest::rstest;
739
740    use super::*;
741
742    fn sample_identity() -> OrderIdentity {
743        OrderIdentity {
744            instrument_id: InstrumentId::from("ETH-PERP.DERIVE"),
745            strategy_id: StrategyId::from("S-1"),
746            order_side: OrderSide::Buy,
747            order_type: OrderType::Limit,
748        }
749    }
750
751    #[rstest]
752    fn test_instrument_precision_roundtrip() {
753        let state = WsDispatchState::new(42);
754        let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
755
756        assert_eq!(state.instrument_precision(&instrument_id), None);
757        state.register_instrument_precision(instrument_id, 2, 3);
758        assert_eq!(state.instrument_precision(&instrument_id), Some((2, 3)));
759    }
760
761    #[rstest]
762    fn test_register_and_identity_roundtrip() {
763        let state = WsDispatchState::new(42);
764        let cid = ClientOrderId::from("STRAT-O-1");
765        let identity = sample_identity();
766
767        assert!(state.identity(&cid).is_none());
768        state.register_identity(cid, identity);
769        assert_eq!(state.identity(&cid), Some(identity));
770
771        state.forget(&cid);
772        assert!(state.identity(&cid).is_none());
773    }
774
775    #[rstest]
776    fn test_mark_accepted_dedupes_second_call() {
777        let state = WsDispatchState::new(42);
778        let cid = ClientOrderId::from("STRAT-O-1");
779
780        assert!(!state.mark_accepted(cid));
781        assert!(state.contains_accepted(&cid));
782        assert!(state.mark_accepted(cid));
783    }
784
785    #[rstest]
786    fn test_active_acceptance_survives_replay_cache_eviction() {
787        let state = WsDispatchState::new(42);
788        let client_order_id = ClientOrderId::from("ACTIVE-0");
789        state.register_identity(client_order_id, sample_identity());
790        state.mark_accepted(client_order_id);
791
792        for n in 1..=ORDER_DEDUP_CAPACITY {
793            let other = ClientOrderId::new(format!("ACTIVE-{n}"));
794            state.register_identity(other, sample_identity());
795            state.mark_accepted(other);
796        }
797
798        assert_eq!(state.identity(&client_order_id), Some(sample_identity()));
799        assert!(state.contains_accepted(&client_order_id));
800        assert!(state.mark_accepted(client_order_id));
801    }
802
803    #[rstest]
804    fn test_check_and_insert_trade_returns_true_on_duplicate() {
805        let state = WsDispatchState::new(42);
806        let trade_id = TradeId::new("T-1");
807
808        assert!(!state.check_and_insert_trade(trade_id));
809        assert!(state.contains_trade(&trade_id));
810        assert!(state.check_and_insert_trade(trade_id));
811    }
812
813    #[rstest]
814    fn test_forget_clears_accepted_marker() {
815        let state = WsDispatchState::new(42);
816        let cid = ClientOrderId::from("STRAT-O-1");
817
818        state.mark_accepted(cid);
819        state.forget(&cid);
820        assert!(!state.contains_accepted(&cid));
821    }
822
823    #[rstest]
824    fn test_bound_venue_order_id_records_and_advances() {
825        let state = WsDispatchState::new(42);
826        let cid = ClientOrderId::from("STRAT-O-1");
827        let voi1 = VenueOrderId::from("voi-1");
828        let voi2 = VenueOrderId::from("voi-2");
829
830        assert!(state.bound_venue_order_id(&cid).is_none());
831        state.record_venue_order_id(cid, voi1);
832        assert_eq!(state.bound_venue_order_id(&cid), Some(voi1));
833        // A modify rebinds the order to the replacement venue order id.
834        state.record_venue_order_id(cid, voi2);
835        assert_eq!(state.bound_venue_order_id(&cid), Some(voi2));
836    }
837
838    #[rstest]
839    fn test_pending_modify_admission_does_not_replace_existing_operation() {
840        let state = WsDispatchState::new(42);
841        let cid = ClientOrderId::from("MODIFY-SERIAL");
842        let old_id = VenueOrderId::from("original-leg");
843        state.mark_pending_modify(cid, old_id);
844        state.record_modify_target(cid, Quantity::from("2.000"), Some(Price::from("3505.00")));
845        state.mark_pending_modify(cid, VenueOrderId::from("overlapping-leg"));
846        assert_eq!(state.pending_modify(&cid), Some(old_id));
847        assert_eq!(
848            state.modify_target(&cid),
849            Some((Quantity::from("2.000"), Some(Price::from("3505.00"))))
850        );
851    }
852
853    #[rstest]
854    fn test_pending_modify_marker_set_and_cleared() {
855        let state = WsDispatchState::new(42);
856        let cid = ClientOrderId::from("STRAT-O-1");
857        let old_voi = VenueOrderId::from("voi-1");
858
859        assert!(state.pending_modify(&cid).is_none());
860        state.mark_pending_modify(cid, old_voi);
861        assert_eq!(state.pending_modify(&cid), Some(old_voi));
862        state.clear_pending_modify(&cid);
863        assert!(state.pending_modify(&cid).is_none());
864    }
865
866    #[rstest]
867    fn test_bind_incoming_modify_advances_bound_venue_order_id() {
868        let state = WsDispatchState::new(42);
869        let cid = ClientOrderId::from("STRAT-O-1");
870        let old_voi = VenueOrderId::from("voi-old");
871        let new_voi = VenueOrderId::from("voi-new");
872        state.record_venue_order_id(cid, old_voi);
873        state.mark_pending_modify(cid, old_voi);
874
875        assert!(state.bind_incoming_modify(cid, new_voi, false));
876        assert_eq!(state.bound_venue_order_id(&cid), Some(new_voi));
877        assert!(!state.bind_incoming_modify(cid, VenueOrderId::from("voi-other"), true));
878        assert_eq!(state.bound_venue_order_id(&cid), Some(new_voi));
879        assert!(state.take_pending_modify(&cid, old_voi, None));
880        assert!(!state.take_pending_modify(&cid, old_voi, None));
881    }
882
883    #[rstest]
884    fn test_terminal_incoming_modify_claims_pending_response() {
885        let state = WsDispatchState::new(42);
886        let cid = ClientOrderId::from("STRAT-O-1");
887        let old_voi = VenueOrderId::from("voi-old");
888        state.record_venue_order_id(cid, old_voi);
889        state.mark_pending_modify(cid, old_voi);
890
891        assert!(state.bind_incoming_modify(cid, VenueOrderId::from("voi-rejected"), true));
892        assert!(state.pending_modify(&cid).is_none());
893        assert!(!state.take_pending_modify(&cid, old_voi, None));
894    }
895
896    #[rstest]
897    fn test_response_claim_rebinds_before_clearing_pending_modify() {
898        let state = WsDispatchState::new(42);
899        let cid = ClientOrderId::from("STRAT-O-1");
900        let old_voi = VenueOrderId::from("voi-old");
901        let new_voi = VenueOrderId::from("voi-new");
902        state.record_venue_order_id(cid, old_voi);
903        state.mark_pending_modify(cid, old_voi);
904
905        assert!(state.take_pending_modify(&cid, old_voi, Some(new_voi)));
906        assert_eq!(state.bound_venue_order_id(&cid), Some(new_voi));
907        assert!(state.pending_modify(&cid).is_none());
908    }
909
910    #[rstest]
911    fn test_forget_clears_bound_and_pending() {
912        let state = WsDispatchState::new(42);
913        let cid = ClientOrderId::from("STRAT-O-1");
914
915        let identity = OrderIdentity {
916            instrument_id: InstrumentId::from("ETH-PERP.DERIVE"),
917            strategy_id: StrategyId::from("S-1"),
918            order_side: OrderSide::Buy,
919            order_type: OrderType::Limit,
920        };
921
922        state.register_identity(cid, identity);
923        state.record_venue_order_id(cid, VenueOrderId::from("voi-0"));
924        state.record_venue_order_id(cid, VenueOrderId::from("voi-1"));
925        state.mark_pending_modify(cid, VenueOrderId::from("voi-1"));
926        state.forget(&cid);
927        assert!(state.bound_venue_order_id(&cid).is_none());
928        assert!(state.pending_modify(&cid).is_none());
929        assert_eq!(state.identity(&cid), None);
930        assert_eq!(state.known_instrument(&cid), Some(identity.instrument_id));
931        assert!(state.is_terminal(&cid, identity.instrument_id));
932        assert_eq!(
933            state.order_binding(&cid),
934            Some(OrderBinding {
935                identity,
936                venue_order_id: Some(VenueOrderId::from("voi-1")),
937                venue_order_legs: vec![VenueOrderId::from("voi-0"), VenueOrderId::from("voi-1")]
938            })
939        );
940    }
941}
nautilus_derive/websocket/dispatch.rs

nautilus_derive/websocket/
dispatch.rs
