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

nautilus_derive/websocket/
dispatch.rs
