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 orders submitted through
19//! this client, and the cross-stream deduplication gates that keep replay
20//! frames and concurrent `.orders` / `.trades` updates from emitting duplicate
21//! events.
22//!
23//! Tracked orders (those whose identity was registered at submission time)
24//! produce proper order events (`OrderAccepted`, `OrderFilled`, `OrderCanceled`,
25//! `OrderExpired`, `OrderRejected`). Untracked frames fall back to execution
26//! reports for downstream reconciliation.
27
28use ahash::AHashMap;
29use nautilus_common::cache::fifo::FifoCache;
30use nautilus_model::{
31    enums::{OrderSide, OrderType},
32    identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId, VenueOrderId},
33};
34use parking_lot::Mutex;
35
36/// Capacity for the cross-source trade-id dedup cache. Sized to cover any
37/// reconciliation lookback window plausible for live trading.
38pub const TRADE_DEDUP_CAPACITY: usize = 4_096;
39
40/// Capacity for the per-order accepted / filled dedup caches. Tracks active
41/// and recently-terminal orders so reconnect replays do not re-emit lifecycle
42/// events; need only span the live-stream replay window plus a margin.
43pub const ORDER_DEDUP_CAPACITY: usize = 1_024;
44
45/// Order identity captured at submission time so the dispatch task can build
46/// proper order events without consulting the cache.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub struct OrderIdentity {
49    pub instrument_id: InstrumentId,
50    pub strategy_id: StrategyId,
51    pub order_side: OrderSide,
52    pub order_type: OrderType,
53}
54
55/// Shared dispatch state for the Derive WS execution loop.
56///
57/// Instrument precision populates from instrument definitions and governs
58/// execution report values. `order_identities` populates on successful
59/// `submit_order` and is consulted by both the `.orders` and `.trades`
60/// dispatch paths to decide whether a frame belongs to a tracked or external
61/// order. `pending_modifies` and `bound_venue_order_ids` track the in-flight
62/// and current venue order id of a `private/replace` so the dispatch suppresses
63/// events for the superseded leg.
64#[derive(Debug, Default)]
65pub struct WsDispatchState {
66    order_identities: Mutex<AHashMap<ClientOrderId, OrderIdentity>>,
67    instrument_precisions: Mutex<AHashMap<InstrumentId, (u8, u8)>>,
68    emitted_accepted: Mutex<FifoCache<ClientOrderId, ORDER_DEDUP_CAPACITY>>,
69    emitted_canceled: Mutex<FifoCache<ClientOrderId, ORDER_DEDUP_CAPACITY>>,
70    filled_orders: Mutex<FifoCache<ClientOrderId, ORDER_DEDUP_CAPACITY>>,
71    emitted_trades: Mutex<FifoCache<TradeId, TRADE_DEDUP_CAPACITY>>,
72    bound_venue_order_ids: Mutex<AHashMap<ClientOrderId, VenueOrderId>>,
73    pending_modifies: Mutex<AHashMap<ClientOrderId, VenueOrderId>>,
74}
75
76impl WsDispatchState {
77    #[must_use]
78    pub fn new() -> Self {
79        Self::default()
80    }
81
82    /// Records price and size precision for execution report parsing.
83    pub(crate) fn register_instrument_precision(
84        &self,
85        instrument_id: InstrumentId,
86        price_precision: u8,
87        size_precision: u8,
88    ) {
89        self.instrument_precisions
90            .lock()
91            .insert(instrument_id, (price_precision, size_precision));
92    }
93
94    /// Returns price and size precision for an instrument, when registered.
95    #[must_use]
96    pub(crate) fn instrument_precision(&self, instrument_id: &InstrumentId) -> Option<(u8, u8)> {
97        self.instrument_precisions
98            .lock()
99            .get(instrument_id)
100            .copied()
101    }
102
103    /// Registers an order identity captured at submission so subsequent WS
104    /// frames for the same client_order_id resolve to the tracked path.
105    pub fn register_identity(&self, client_order_id: ClientOrderId, identity: OrderIdentity) {
106        self.order_identities
107            .lock()
108            .insert(client_order_id, identity);
109    }
110
111    /// Returns the registered identity for a client order, when one was
112    /// captured at submission time.
113    #[must_use]
114    pub fn identity(&self, client_order_id: &ClientOrderId) -> Option<OrderIdentity> {
115        self.order_identities.lock().get(client_order_id).copied()
116    }
117
118    /// Drops identity and the accepted marker for a terminal order so future
119    /// stale frames (post-cancel cleanup, history backfill) take the untracked
120    /// report path.
121    pub fn forget(&self, client_order_id: &ClientOrderId) {
122        self.order_identities.lock().remove(client_order_id);
123        self.emitted_accepted.lock().remove(client_order_id);
124        self.bound_venue_order_ids.lock().remove(client_order_id);
125        self.pending_modifies.lock().remove(client_order_id);
126    }
127
128    /// Records the venue order id currently bound to a tracked client order.
129    pub fn record_venue_order_id(
130        &self,
131        client_order_id: ClientOrderId,
132        venue_order_id: VenueOrderId,
133    ) {
134        self.bound_venue_order_ids
135            .lock()
136            .insert(client_order_id, venue_order_id);
137    }
138
139    /// Returns the venue order id currently bound to a tracked client order.
140    #[must_use]
141    pub fn bound_venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
142        self.bound_venue_order_ids
143            .lock()
144            .get(client_order_id)
145            .copied()
146    }
147
148    /// Records the old venue order id of an in-flight `private/replace`, set
149    /// before the request so the cancel leg is suppressed.
150    pub fn mark_pending_modify(
151        &self,
152        client_order_id: ClientOrderId,
153        old_venue_order_id: VenueOrderId,
154    ) {
155        self.pending_modifies
156            .lock()
157            .insert(client_order_id, old_venue_order_id);
158    }
159
160    /// Clears the in-flight modify marker once the replace resolves.
161    pub fn clear_pending_modify(&self, client_order_id: &ClientOrderId) {
162        self.pending_modifies.lock().remove(client_order_id);
163    }
164
165    /// Returns the old venue order id of an in-flight modify, when one is set.
166    #[must_use]
167    pub fn pending_modify(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
168        self.pending_modifies.lock().get(client_order_id).copied()
169    }
170
171    /// Rebinds an in-flight modify when a frame for its replacement arrives
172    /// before the `private/replace` response.
173    pub fn bind_incoming_modify(
174        &self,
175        client_order_id: ClientOrderId,
176        venue_order_id: VenueOrderId,
177        terminal: bool,
178    ) -> bool {
179        let mut pending = self.pending_modifies.lock();
180        let Some(old_venue_order_id) = pending.get(&client_order_id).copied() else {
181            return false;
182        };
183
184        if old_venue_order_id == venue_order_id {
185            return false;
186        }
187
188        let mut bound = self.bound_venue_order_ids.lock();
189        if bound
190            .get(&client_order_id)
191            .is_some_and(|current| *current != old_venue_order_id)
192        {
193            return false;
194        }
195        bound.insert(client_order_id, venue_order_id);
196        if terminal {
197            pending.remove(&client_order_id);
198        }
199        true
200    }
201
202    /// Atomically claims a pending modify for its RPC response, optionally
203    /// rebinding the replacement venue order id before clearing the marker.
204    /// Returns `false` when an incoming terminal frame already resolved it.
205    pub fn take_pending_modify(
206        &self,
207        client_order_id: &ClientOrderId,
208        old_venue_order_id: VenueOrderId,
209        new_venue_order_id: Option<VenueOrderId>,
210    ) -> bool {
211        let mut pending = self.pending_modifies.lock();
212        if pending.get(client_order_id) != Some(&old_venue_order_id) {
213            return false;
214        }
215
216        if let Some(new_venue_order_id) = new_venue_order_id {
217            self.bound_venue_order_ids
218                .lock()
219                .insert(*client_order_id, new_venue_order_id);
220        }
221        pending.remove(client_order_id);
222        true
223    }
224
225    /// Returns `true` when an `OrderAccepted` has already been emitted for
226    /// this client order in the current process lifetime.
227    #[must_use]
228    pub fn contains_accepted(&self, client_order_id: &ClientOrderId) -> bool {
229        self.emitted_accepted.lock().contains(client_order_id)
230    }
231
232    /// Records that `OrderAccepted` has been emitted for this client order.
233    /// Returns `true` when the marker was already present (duplicate).
234    pub fn mark_accepted(&self, client_order_id: ClientOrderId) -> bool {
235        let mut cache = self.emitted_accepted.lock();
236        if cache.contains(&client_order_id) {
237            return true;
238        }
239        cache.add(client_order_id);
240        false
241    }
242
243    /// Records that `OrderCanceled` has been emitted for this client order.
244    /// Returns `true` when the marker was already present (duplicate).
245    pub fn mark_canceled(&self, client_order_id: ClientOrderId) -> bool {
246        let mut cache = self.emitted_canceled.lock();
247        if cache.contains(&client_order_id) {
248            return true;
249        }
250        cache.add(client_order_id);
251        false
252    }
253
254    /// Returns `true` when this client order has reached a terminal filled
255    /// state, used to suppress stale Accepted frames replayed on reconnect.
256    #[must_use]
257    pub fn contains_filled(&self, client_order_id: &ClientOrderId) -> bool {
258        self.filled_orders.lock().contains(client_order_id)
259    }
260
261    /// Marks the client order as terminally filled. Idempotent.
262    pub fn mark_filled(&self, client_order_id: ClientOrderId) {
263        let mut cache = self.filled_orders.lock();
264        if !cache.contains(&client_order_id) {
265            cache.add(client_order_id);
266        }
267    }
268
269    /// Inserts the trade id atomically. Returns `true` when the id was
270    /// already present (i.e., this fill should be skipped as a duplicate).
271    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
272        let mut cache = self.emitted_trades.lock();
273        if cache.contains(&trade_id) {
274            return true;
275        }
276        cache.add(trade_id);
277        false
278    }
279
280    /// Returns `true` when this trade id has already been seen, without
281    /// mutating state.
282    #[must_use]
283    pub fn contains_trade(&self, trade_id: &TradeId) -> bool {
284        self.emitted_trades.lock().contains(trade_id)
285    }
286}
287
288#[cfg(test)]
289mod tests {
290    use nautilus_model::{
291        enums::{OrderSide, OrderType},
292        identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId, VenueOrderId},
293    };
294    use rstest::rstest;
295
296    use super::*;
297
298    fn sample_identity() -> OrderIdentity {
299        OrderIdentity {
300            instrument_id: InstrumentId::from("ETH-PERP.DERIVE"),
301            strategy_id: StrategyId::from("S-1"),
302            order_side: OrderSide::Buy,
303            order_type: OrderType::Limit,
304        }
305    }
306
307    #[rstest]
308    fn test_instrument_precision_roundtrip() {
309        let state = WsDispatchState::new();
310        let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
311
312        assert_eq!(state.instrument_precision(&instrument_id), None);
313        state.register_instrument_precision(instrument_id, 2, 3);
314        assert_eq!(state.instrument_precision(&instrument_id), Some((2, 3)));
315    }
316
317    #[rstest]
318    fn test_register_and_identity_roundtrip() {
319        let state = WsDispatchState::new();
320        let cid = ClientOrderId::from("STRAT-O-1");
321        let identity = sample_identity();
322
323        assert!(state.identity(&cid).is_none());
324        state.register_identity(cid, identity);
325        assert_eq!(state.identity(&cid), Some(identity));
326
327        state.forget(&cid);
328        assert!(state.identity(&cid).is_none());
329    }
330
331    #[rstest]
332    fn test_mark_accepted_dedupes_second_call() {
333        let state = WsDispatchState::new();
334        let cid = ClientOrderId::from("STRAT-O-1");
335
336        assert!(!state.mark_accepted(cid));
337        assert!(state.contains_accepted(&cid));
338        assert!(state.mark_accepted(cid));
339    }
340
341    #[rstest]
342    fn test_check_and_insert_trade_returns_true_on_duplicate() {
343        let state = WsDispatchState::new();
344        let trade_id = TradeId::new("T-1");
345
346        assert!(!state.check_and_insert_trade(trade_id));
347        assert!(state.contains_trade(&trade_id));
348        assert!(state.check_and_insert_trade(trade_id));
349    }
350
351    #[rstest]
352    fn test_forget_clears_accepted_marker() {
353        let state = WsDispatchState::new();
354        let cid = ClientOrderId::from("STRAT-O-1");
355
356        state.mark_accepted(cid);
357        state.forget(&cid);
358        assert!(!state.contains_accepted(&cid));
359    }
360
361    #[rstest]
362    fn test_bound_venue_order_id_records_and_advances() {
363        let state = WsDispatchState::new();
364        let cid = ClientOrderId::from("STRAT-O-1");
365        let voi1 = VenueOrderId::from("voi-1");
366        let voi2 = VenueOrderId::from("voi-2");
367
368        assert!(state.bound_venue_order_id(&cid).is_none());
369        state.record_venue_order_id(cid, voi1);
370        assert_eq!(state.bound_venue_order_id(&cid), Some(voi1));
371        // A modify rebinds the order to the replacement venue order id.
372        state.record_venue_order_id(cid, voi2);
373        assert_eq!(state.bound_venue_order_id(&cid), Some(voi2));
374    }
375
376    #[rstest]
377    fn test_pending_modify_marker_set_and_cleared() {
378        let state = WsDispatchState::new();
379        let cid = ClientOrderId::from("STRAT-O-1");
380        let old_voi = VenueOrderId::from("voi-1");
381
382        assert!(state.pending_modify(&cid).is_none());
383        state.mark_pending_modify(cid, old_voi);
384        assert_eq!(state.pending_modify(&cid), Some(old_voi));
385        state.clear_pending_modify(&cid);
386        assert!(state.pending_modify(&cid).is_none());
387    }
388
389    #[rstest]
390    fn test_bind_incoming_modify_advances_bound_venue_order_id() {
391        let state = WsDispatchState::new();
392        let cid = ClientOrderId::from("STRAT-O-1");
393        let old_voi = VenueOrderId::from("voi-old");
394        let new_voi = VenueOrderId::from("voi-new");
395        state.record_venue_order_id(cid, old_voi);
396        state.mark_pending_modify(cid, old_voi);
397
398        assert!(state.bind_incoming_modify(cid, new_voi, false));
399        assert_eq!(state.bound_venue_order_id(&cid), Some(new_voi));
400        assert!(!state.bind_incoming_modify(cid, VenueOrderId::from("voi-other"), true));
401        assert_eq!(state.bound_venue_order_id(&cid), Some(new_voi));
402        assert!(state.take_pending_modify(&cid, old_voi, None));
403        assert!(!state.take_pending_modify(&cid, old_voi, None));
404    }
405
406    #[rstest]
407    fn test_terminal_incoming_modify_claims_pending_response() {
408        let state = WsDispatchState::new();
409        let cid = ClientOrderId::from("STRAT-O-1");
410        let old_voi = VenueOrderId::from("voi-old");
411        state.record_venue_order_id(cid, old_voi);
412        state.mark_pending_modify(cid, old_voi);
413
414        assert!(state.bind_incoming_modify(cid, VenueOrderId::from("voi-rejected"), true));
415        assert!(state.pending_modify(&cid).is_none());
416        assert!(!state.take_pending_modify(&cid, old_voi, None));
417    }
418
419    #[rstest]
420    fn test_response_claim_rebinds_before_clearing_pending_modify() {
421        let state = WsDispatchState::new();
422        let cid = ClientOrderId::from("STRAT-O-1");
423        let old_voi = VenueOrderId::from("voi-old");
424        let new_voi = VenueOrderId::from("voi-new");
425        state.record_venue_order_id(cid, old_voi);
426        state.mark_pending_modify(cid, old_voi);
427
428        assert!(state.take_pending_modify(&cid, old_voi, Some(new_voi)));
429        assert_eq!(state.bound_venue_order_id(&cid), Some(new_voi));
430        assert!(state.pending_modify(&cid).is_none());
431    }
432
433    #[rstest]
434    fn test_forget_clears_bound_and_pending() {
435        let state = WsDispatchState::new();
436        let cid = ClientOrderId::from("STRAT-O-1");
437
438        state.record_venue_order_id(cid, VenueOrderId::from("voi-1"));
439        state.mark_pending_modify(cid, VenueOrderId::from("voi-1"));
440        state.forget(&cid);
441        assert!(state.bound_venue_order_id(&cid).is_none());
442        assert!(state.pending_modify(&cid).is_none());
443    }
444}
nautilus_derive/websocket/dispatch.rs

nautilus_derive/websocket/
dispatch.rs
