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

nautilus_derive/websocket/
dispatch.rs
