nautilus_hyperliquid/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//! WebSocket execution dispatch for the Hyperliquid execution client.
17//!
18//! Implements the two-tier execution dispatch contract from
19//! `docs/developer_guide/adapters.md#tracked-and-external-execution-updates`:
20//!
21//! 1. The execution client registers an [`OrderContext`] in [`WsDispatchState`]
22//!    when it submits an order, and refreshes the cached venue order id when a
23//!    modify is sent so the WebSocket consumer can detect cancel-replace.
24//! 2. Incoming [`OrderStatusReport`] and [`FillReport`] messages are routed
25//!    through [`dispatch_order_event`] and [`dispatch_order_fill`].
26//!    For tracked orders these build typed [`OrderEventAny`] events and emit
27//!    them via [`ExecutionEventEmitter::send_order_event`]. For untracked /
28//!    external orders the dispatch falls back to forwarding the raw report.
29//!
30//! The dispatch state lives in an `Arc<WsDispatchState>` shared between the
31//! main client task (which registers contexts at submission time) and the
32//! spawned WebSocket consumer task.
33//!
34//! # GH-3827 cancel-replace handling
35//!
36//! Hyperliquid implements `modify` as a cancel-and-replace: the venue emits an
37//! `ACCEPTED(new_voi)` together with a `CANCELED(old_voi)` under the same
38//! `client_order_id`. The dispatch detects the replacement leg by comparing
39//! `report.venue_order_id` to the last cached value, promotes it to an
40//! `OrderUpdated` event, and suppresses the stale cancel so strategies never
41//! observe a spurious termination.
42//!
43//! Each in-flight modify is tracked as an intent in a per-order chain (keyed
44//! on `client_order_id`), pushed by `modify_order` before the HTTP call. An
45//! intent lets dispatch skip an early `CANCELED(old_voi)` that arrives before
46//! the replacement `ACCEPTED(new_voi)`, regardless of whether the WS message
47//! races ahead of the HTTP response. Rapid repeated modifies under one stable
48//! CLOID queue as a chain so a later modify cannot overwrite an earlier
49//! intent's old-leg suppression, and a failed modify clears only its own
50//! generation (leaving newer intents intact). The front intent is claimed on
51//! promotion, advancing the next intent's old leg to the promoted replacement;
52//! a rejected front reparents the next intent to the same still-live leg.
53//!
54//! A fill carrying the replacement `venue_order_id` during an in-flight modify
55//! promotes the binding directly (the same `OrderUpdated` path as the
56//! replacement `ACCEPTED`), so a dropped `ACCEPTED` does not strand the fill.
57//! A fill is buffered into [`WsDispatchState::buffered_fills`] only when the
58//! context has no price to promote with; `handle_accepted` drains the buffer
59//! on the replacement `ACCEPTED`. A delayed earlier-leg fill during a chained
60//! modify is a known limitation. See GH-3972.
61//!
62//! When neither the replacement `ACCEPTED` nor a fill arrives, a query that
63//! resolves the replacement by `cloid` promotes the binding the same way via
64//! [`promote_replacement_from_query`], so a dropped `ACCEPTED` with no fill
65//! cannot leave the order bound to the canceled leg.
66
67use std::{
68    collections::VecDeque,
69    sync::atomic::{AtomicBool, Ordering},
70};
71
72use dashmap::{DashMap, DashSet};
73use nautilus_common::cache::fifo::FifoCache;
74use nautilus_core::{UUID4, UnixNanos};
75use nautilus_live::{ExecutionEventEmitter, execution::context::OrderContext};
76use nautilus_model::{
77    enums::{OrderStatus, OrderType},
78    events::{
79        OrderAccepted, OrderCanceled, OrderEventAny, OrderExpired, OrderFilled, OrderRejected,
80        OrderTriggered, OrderUpdated,
81    },
82    identifiers::{AccountId, ClientOrderId, TradeId, VenueOrderId},
83    reports::{FillReport, OrderStatusReport},
84    types::{Price, Quantity},
85};
86use parking_lot::Mutex;
87use ustr::Ustr;
88
89use crate::{
90    common::consts::HYPERLIQUID_POST_ONLY_WOULD_MATCH,
91    http::models::HyperliquidExchangePlaceOrderRequest,
92};
93
94pub const DEDUP_CAPACITY: usize = 10_000;
95
96/// Maximum in-flight modify intents tracked per order. Rapid repricing rarely
97/// queues more than one or two unacknowledged modifies at once; the cap bounds
98/// memory if replacement acks stall. On overflow the oldest intent is evicted.
99pub const MAX_PENDING_MODIFY_INTENTS: usize = 32;
100
101/// A single in-flight Hyperliquid modify awaiting its replacement leg.
102///
103/// Rapid repeated modifies under one stable CLOID queue as a chain of intents
104/// so a later modify cannot overwrite an earlier pending old-leg marker, and a
105/// failed modify clears only its own generation rather than a newer one's
106/// state. Each intent carries the venue leg it cancel-replaces
107/// (`old_venue_order_id`), the user-intended absolute total quantity, and the
108/// exact request sent (used to size a corrective reduce).
109#[derive(Debug, Clone)]
110pub struct ModifyIntent {
111    /// Monotonic per-order generation, used to clear a specific intent on failure.
112    pub generation: u64,
113    /// Venue order id this modify cancel-replaces, once known.
114    pub old_venue_order_id: Option<VenueOrderId>,
115    /// User-intended absolute total quantity for the replacement.
116    pub target_qty: Quantity,
117    /// Exact venue request sent, used to size a corrective reduce.
118    pub sent_request: Option<HyperliquidExchangePlaceOrderRequest>,
119}
120
121/// Bounded FIFO chain of in-flight modify intents for one order.
122///
123/// The venue processes chained modifies in submission order, so the front
124/// (oldest) intent is the next to promote; on promotion the next intent's old
125/// leg advances to the replacement just accepted.
126#[derive(Debug, Default)]
127struct ModifyChain {
128    intents: VecDeque<ModifyIntent>,
129    next_generation: u64,
130}
131
132/// Per-client dispatch state shared between order submission and the
133/// WebSocket consumer task.
134///
135/// Tracks which orders were submitted through this client (so we can route
136/// venue events to typed [`OrderEventAny`] emissions for tracked orders and
137/// fall back to reports for external orders), provides cross-stream dedup
138/// for `OrderAccepted` and `OrderFilled` emissions, and carries the
139/// GH-3827 cancel-replace state (`cached_venue_order_ids` and
140/// `pending_modify_keys`).
141#[derive(Debug)]
142pub struct WsDispatchState {
143    /// Tracked orders keyed by full Nautilus [`ClientOrderId`].
144    ///
145    /// The dispatch functions read the context to build typed order events for
146    /// tracked orders without needing access to the engine cache (which is
147    /// `!Send` and unreachable from the spawned WebSocket task). `quantity` and
148    /// `price` are refreshed from subsequent status reports so a cancel-replace
149    /// `ACCEPTED` that omits `price` can still produce an `OrderUpdated`
150    /// carrying an accurate value.
151    pub order_contexts: DashMap<ClientOrderId, OrderContext>,
152    /// Client order IDs for which an `OrderAccepted` event has been emitted.
153    pub emitted_accepted: DashSet<ClientOrderId>,
154    /// Tracked submissions whose POST response has not resolved yet.
155    pending_submissions: DashSet<ClientOrderId>,
156    /// Submission-time rejections held until the POST path can preserve its
157    /// more detailed venue error string.
158    pending_submission_rejections: DashMap<ClientOrderId, OrderStatusReport>,
159    /// Client order IDs that have reached the filled terminal state.
160    ///
161    /// Retained past `cleanup_terminal` so that late replay of the same
162    /// status or fill does not re-emit events.
163    pub filled_orders: DashSet<ClientOrderId>,
164    /// Trade IDs for which an `OrderFilled` event has been emitted.
165    ///
166    /// Bounded FIFO dedup to bound memory while keeping recent trade ids
167    /// deduped across reconnects.
168    emitted_trades: Mutex<FifoCache<TradeId, DEDUP_CAPACITY>>,
169    /// Raw Hyperliquid CLOIDs that reached a terminal state through the post
170    /// response path before the matching `orderUpdates` event arrived.
171    terminal_cloids: Mutex<FifoCache<Ustr, DEDUP_CAPACITY>>,
172    /// Last venue order id observed for a tracked client order id.
173    ///
174    /// Populated on the first `OrderAccepted` and refreshed on every
175    /// cancel-replace promotion. A later `ACCEPTED` with a different venue
176    /// order id under the same client order id is treated as the
177    /// replacement leg of a Hyperliquid modify and emitted as `OrderUpdated`.
178    pub cached_venue_order_ids: DashMap<ClientOrderId, VenueOrderId>,
179    /// Per-order chain of in-flight modify intents, keyed by `client_order_id`.
180    ///
181    /// Rapid repeated modifies under one stable CLOID queue as a chain so a
182    /// later modify cannot overwrite an earlier pending old-leg marker, and a
183    /// failed modify clears only its own generation rather than a newer one's
184    /// state. Populated by `modify_order` before the HTTP call so the WS cancel
185    /// handler sees an intent even when `CANCELED(old_voi)` arrives before the
186    /// HTTP response. A `CANCELED(old_voi)` matching any queued intent's old
187    /// leg is suppressed so the later `ACCEPTED(new_voi)` can flow through the
188    /// `OrderUpdated` path; the front intent is claimed on promotion and the
189    /// next intent's old leg advances to the promoted replacement.
190    pending_modify_chains: DashMap<ClientOrderId, ModifyChain>,
191    /// `FillReport`s buffered only when a cancel-replace fill cannot be promoted
192    /// (the context carries no price); drained by the cancel-replace branch of
193    /// `handle_accepted`. The common path promotes on the fill instead. See
194    /// GH-3972.
195    pub buffered_fills: DashMap<ClientOrderId, Vec<FillReport>>,
196    /// Cumulative filled quantity per tracked order. Compared against
197    /// `OrderContext::quantity` to decide when to clean up tracked state.
198    pub order_filled_qty: DashMap<ClientOrderId, Quantity>,
199    /// Corrective reduce queued by the cancel-replace promotion: client order
200    /// id to (new venue order id, reduced request). Drained by the WS loop.
201    pub pending_corrective: DashMap<ClientOrderId, (u64, HyperliquidExchangePlaceOrderRequest)>,
202    clearing: AtomicBool,
203}
204
205impl Default for WsDispatchState {
206    fn default() -> Self {
207        Self {
208            order_contexts: DashMap::new(),
209            emitted_accepted: DashSet::default(),
210            pending_submissions: DashSet::default(),
211            pending_submission_rejections: DashMap::new(),
212            filled_orders: DashSet::default(),
213            emitted_trades: Mutex::new(FifoCache::new()),
214            terminal_cloids: Mutex::new(FifoCache::new()),
215            cached_venue_order_ids: DashMap::new(),
216            pending_modify_chains: DashMap::new(),
217            buffered_fills: DashMap::new(),
218            order_filled_qty: DashMap::new(),
219            pending_corrective: DashMap::new(),
220            clearing: AtomicBool::new(false),
221        }
222    }
223}
224
225impl WsDispatchState {
226    /// Creates a new empty dispatch state.
227    #[must_use]
228    pub fn new() -> Self {
229        Self::default()
230    }
231
232    /// Registers an order context. Called by the execution client at order
233    /// submission time, before any WebSocket events for the order can arrive.
234    pub fn register_context(&self, context: OrderContext) {
235        self.order_contexts
236            .insert(context.identity.client_order_id, context);
237    }
238
239    /// Returns a copy of the context for the given client order id, if any.
240    #[must_use]
241    pub fn lookup_context(&self, client_order_id: &ClientOrderId) -> Option<OrderContext> {
242        self.order_contexts.get(client_order_id).map(|r| *r)
243    }
244
245    /// Marks a tracked order as awaiting its submission POST response.
246    pub fn mark_submission_pending(&self, client_order_id: ClientOrderId) {
247        self.pending_submissions.insert(client_order_id);
248    }
249
250    /// Returns whether the order still awaits its submission POST response.
251    #[must_use]
252    pub fn submission_pending(&self, client_order_id: &ClientOrderId) -> bool {
253        self.pending_submissions.contains(client_order_id)
254    }
255
256    /// Holds a submission-time rejection until the POST response resolves.
257    pub fn buffer_submission_rejection(
258        &self,
259        client_order_id: ClientOrderId,
260        report: OrderStatusReport,
261    ) {
262        self.pending_submission_rejections
263            .insert(client_order_id, report);
264    }
265
266    /// Resolves submission tracking and returns any early rejection report.
267    #[must_use]
268    pub fn resolve_submission(&self, client_order_id: &ClientOrderId) -> Option<OrderStatusReport> {
269        self.pending_submissions.remove(client_order_id);
270        self.pending_submission_rejections
271            .remove(client_order_id)
272            .map(|(_, report)| report)
273    }
274
275    /// Refreshes the tracked price for a modify ack when the new report
276    /// carries an updated price.
277    pub fn update_context_price(&self, client_order_id: &ClientOrderId, price: Option<Price>) {
278        if let Some(price) = price
279            && let Some(mut entry) = self.order_contexts.get_mut(client_order_id)
280        {
281            entry.price = Some(price);
282        }
283    }
284
285    /// Refreshes the tracked quantity for a modify ack.
286    pub fn update_context_quantity(&self, client_order_id: &ClientOrderId, quantity: Quantity) {
287        if let Some(mut entry) = self.order_contexts.get_mut(client_order_id) {
288            entry.quantity = quantity;
289        }
290    }
291
292    /// Marks an `OrderAccepted` event as emitted for this order.
293    pub fn insert_accepted(&self, cid: ClientOrderId) {
294        self.evict_if_full(&self.emitted_accepted);
295        self.emitted_accepted.insert(cid);
296    }
297
298    /// Marks an order as having reached a terminal state.
299    ///
300    /// Returns `true` when this call claimed the terminal state, and `false`
301    /// when another path had already claimed it.
302    pub fn insert_filled(&self, cid: ClientOrderId) -> bool {
303        self.evict_if_full(&self.filled_orders);
304        self.filled_orders.insert(cid)
305    }
306
307    /// Atomically inserts a trade id into the dedup set.
308    ///
309    /// Returns `true` when the trade was already present (i.e. it is a
310    /// duplicate), `false` otherwise.
311    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
312        let mut set = self.emitted_trades.lock();
313        !set.insert(trade_id)
314    }
315
316    /// Records a terminal raw Hyperliquid CLOID.
317    ///
318    /// Used when the post response rejects an order before the WebSocket
319    /// `orderUpdates` message. The normal CLOID mapping can be removed while a
320    /// late unresolved order update still gets suppressed instead of forwarded
321    /// as an external report.
322    pub fn insert_terminal_cloid(&self, cloid: Ustr) {
323        let mut set = self.terminal_cloids.lock();
324        let _ = set.insert(cloid);
325    }
326
327    /// Returns whether a raw Hyperliquid CLOID reached a terminal state through
328    /// the post response path.
329    #[must_use]
330    pub fn terminal_cloid_seen(&self, cloid: &Ustr) -> bool {
331        let set = self.terminal_cloids.lock();
332        set.contains(cloid)
333    }
334
335    /// Caches the venue order id observed for a tracked client order id.
336    pub fn record_venue_order_id(
337        &self,
338        client_order_id: ClientOrderId,
339        venue_order_id: VenueOrderId,
340    ) {
341        self.cached_venue_order_ids
342            .insert(client_order_id, venue_order_id);
343    }
344
345    /// Returns the previously cached venue order id, if any.
346    #[must_use]
347    pub fn cached_venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
348        self.cached_venue_order_ids.get(client_order_id).map(|r| *r)
349    }
350
351    /// Queues an in-flight modify intent for cancel-before-accept suppression
352    /// and records the target absolute total qty for the cancel-replace
353    /// promotion. Returns the intent's generation.
354    ///
355    /// The generation lets the submission path clear only this modify on
356    /// failure via [`Self::clear_modify_generation`], leaving newer queued
357    /// modifies intact. Chained modifies append rather than overwrite, so a
358    /// later modify cannot drop an earlier pending old-leg marker.
359    pub fn mark_pending_modify(
360        &self,
361        client_order_id: ClientOrderId,
362        old_venue_order_id: VenueOrderId,
363        target_qty: Quantity,
364    ) -> u64 {
365        let mut chain = self
366            .pending_modify_chains
367            .entry(client_order_id)
368            .or_default();
369        let generation = chain.next_generation;
370        chain.next_generation += 1;
371        chain.intents.push_back(ModifyIntent {
372            generation,
373            old_venue_order_id: Some(old_venue_order_id),
374            target_qty,
375            sent_request: None,
376        });
377
378        if chain.intents.len() > MAX_PENDING_MODIFY_INTENTS {
379            chain.intents.pop_front();
380            log::warn!(
381                "Modify chain for {client_order_id} exceeded {MAX_PENDING_MODIFY_INTENTS}; \
382                 evicting oldest intent",
383            );
384        }
385        generation
386    }
387
388    /// Clears the entire pending modify chain for a client order id.
389    pub fn clear_pending_modify(&self, client_order_id: &ClientOrderId) {
390        self.pending_modify_chains.remove(client_order_id);
391    }
392
393    /// Removes a single modify intent by generation, leaving newer queued
394    /// modifies intact. Drops the chain entry when it empties.
395    ///
396    /// When the removed intent is the front, the next queued modify inherits
397    /// its old leg: a rejected modify does not cancel-replace, so the resting
398    /// leg it targeted is still live and the next modify cancel-replaces the
399    /// same one. A non-front removal needs no reparenting; the front's
400    /// promotion (or its own removal) advances the chain.
401    pub fn clear_modify_generation(&self, client_order_id: &ClientOrderId, generation: u64) {
402        let Some(mut chain) = self.pending_modify_chains.get_mut(client_order_id) else {
403            return;
404        };
405        let removed_front_old = chain
406            .intents
407            .front()
408            .filter(|front| front.generation == generation)
409            .and_then(|front| front.old_venue_order_id);
410        chain
411            .intents
412            .retain(|intent| intent.generation != generation);
413
414        if let Some(old) = removed_front_old
415            && let Some(new_front) = chain.intents.front_mut()
416        {
417            new_front.old_venue_order_id = Some(old);
418        }
419        drop(chain);
420        // Remove only if still empty: a concurrent mark for the same order may
421        // queue a new intent between the drop above and this remove
422        self.pending_modify_chains
423            .remove_if(client_order_id, |_, chain| chain.intents.is_empty());
424    }
425
426    /// Stashes the exact venue request sent onto the most recently queued
427    /// modify intent for the order.
428    pub fn stash_modify_request(
429        &self,
430        client_order_id: ClientOrderId,
431        request: HyperliquidExchangePlaceOrderRequest,
432    ) {
433        if let Some(mut chain) = self.pending_modify_chains.get_mut(&client_order_id)
434            && let Some(back) = chain.intents.back_mut()
435        {
436            back.sent_request = Some(request);
437        } else {
438            log::debug!(
439                "Stash modify request for {client_order_id} with no pending intent; ignoring"
440            );
441        }
442    }
443
444    /// Returns a clone of the front intent's stashed modify request, if any.
445    #[must_use]
446    pub fn modify_request(
447        &self,
448        client_order_id: &ClientOrderId,
449    ) -> Option<HyperliquidExchangePlaceOrderRequest> {
450        self.pending_modify_chains
451            .get(client_order_id)
452            .and_then(|chain| chain.intents.front().and_then(|i| i.sent_request.clone()))
453    }
454
455    /// Claims the front (oldest) modify intent for promotion.
456    ///
457    /// Advances the next queued intent's old leg to `new_venue_order_id`: its
458    /// cancel-replace targets the replacement just promoted, not the leg it was
459    /// queued against. Returns the claimed intent, or `None` when no intent is
460    /// queued (an external modify with no local marker). Drops the chain entry
461    /// when it empties.
462    pub fn claim_front_modify(
463        &self,
464        client_order_id: &ClientOrderId,
465        new_venue_order_id: VenueOrderId,
466    ) -> Option<ModifyIntent> {
467        let mut chain = self.pending_modify_chains.get_mut(client_order_id)?;
468        let claimed = chain.intents.pop_front();
469        if let Some(next) = chain.intents.front_mut() {
470            next.old_venue_order_id = Some(new_venue_order_id);
471        }
472        drop(chain);
473        // Remove only if still empty: a concurrent mark for the same order may
474        // queue a new intent between the drop above and this remove
475        self.pending_modify_chains
476            .remove_if(client_order_id, |_, chain| chain.intents.is_empty());
477        claimed
478    }
479
480    /// Queues a corrective reduce for the WebSocket consumer loop to post.
481    pub fn queue_corrective(
482        &self,
483        client_order_id: ClientOrderId,
484        oid: u64,
485        request: HyperliquidExchangePlaceOrderRequest,
486    ) {
487        self.pending_corrective
488            .insert(client_order_id, (oid, request));
489    }
490
491    /// Removes and returns a queued corrective reduce, if any.
492    #[must_use]
493    pub fn take_corrective(
494        &self,
495        client_order_id: &ClientOrderId,
496    ) -> Option<(u64, HyperliquidExchangePlaceOrderRequest)> {
497        self.pending_corrective
498            .remove(client_order_id)
499            .map(|(_, v)| v)
500    }
501
502    /// Returns whether any modify intent is queued for the client order id.
503    #[must_use]
504    pub fn has_pending_modify(&self, client_order_id: &ClientOrderId) -> bool {
505        self.pending_modify_chains
506            .get(client_order_id)
507            .is_some_and(|chain| !chain.intents.is_empty())
508    }
509
510    /// Returns the front intent's old venue order id, if any.
511    #[must_use]
512    pub fn pending_modify(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
513        self.pending_modify_chains
514            .get(client_order_id)
515            .and_then(|chain| chain.intents.front().and_then(|i| i.old_venue_order_id))
516    }
517
518    /// Returns whether any queued intent cancel-replaces `venue_order_id`.
519    ///
520    /// Used to suppress the `CANCELED(old_voi)` leg of any in-flight modify in
521    /// the chain, not only the oldest.
522    #[must_use]
523    pub fn pending_modify_contains_old(
524        &self,
525        client_order_id: &ClientOrderId,
526        venue_order_id: VenueOrderId,
527    ) -> bool {
528        self.pending_modify_chains
529            .get(client_order_id)
530            .is_some_and(|chain| {
531                chain
532                    .intents
533                    .iter()
534                    .any(|i| i.old_venue_order_id == Some(venue_order_id))
535            })
536    }
537
538    /// Returns the front intent's recorded target absolute total qty, if any.
539    #[must_use]
540    pub fn pending_modify_target_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
541        self.pending_modify_chains
542            .get(client_order_id)
543            .and_then(|chain| chain.intents.front().map(|i| i.target_qty))
544    }
545
546    /// Buffers a `FillReport` arrived during an in-flight cancel-replace.
547    pub fn buffer_fill(&self, client_order_id: ClientOrderId, fill: FillReport) {
548        self.buffered_fills
549            .entry(client_order_id)
550            .or_default()
551            .push(fill);
552    }
553
554    /// Removes and returns buffered fills for the cid, in arrival order.
555    #[must_use]
556    pub fn drain_buffered_fills(&self, client_order_id: &ClientOrderId) -> Vec<FillReport> {
557        self.buffered_fills
558            .remove(client_order_id)
559            .map(|(_, v)| v)
560            .unwrap_or_default()
561    }
562
563    /// Number of buffered fills for the cid.
564    #[must_use]
565    pub fn buffered_fill_count(&self, client_order_id: &ClientOrderId) -> usize {
566        self.buffered_fills
567            .get(client_order_id)
568            .map_or(0, |r| r.len())
569    }
570
571    /// Records cumulative filled quantity for a tracked order.
572    pub fn record_filled_qty(&self, client_order_id: ClientOrderId, qty: Quantity) {
573        self.order_filled_qty.insert(client_order_id, qty);
574    }
575
576    /// Returns the previously recorded cumulative filled quantity, if any.
577    #[must_use]
578    pub fn previous_filled_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
579        self.order_filled_qty.get(client_order_id).map(|r| *r)
580    }
581
582    /// Removes all dispatch state for an order that has reached a terminal state.
583    ///
584    /// `filled_orders` is intentionally *not* cleared here: the marker is
585    /// used to suppress stale replays and must outlive the context cleanup.
586    pub fn cleanup_terminal(&self, client_order_id: &ClientOrderId) {
587        self.order_contexts.remove(client_order_id);
588        self.emitted_accepted.remove(client_order_id);
589        self.pending_submissions.remove(client_order_id);
590        self.pending_submission_rejections.remove(client_order_id);
591        self.cached_venue_order_ids.remove(client_order_id);
592        self.pending_modify_chains.remove(client_order_id);
593        self.pending_corrective.remove(client_order_id);
594        self.buffered_fills.remove(client_order_id);
595        self.order_filled_qty.remove(client_order_id);
596    }
597
598    fn evict_if_full(&self, set: &DashSet<ClientOrderId>) {
599        if set.len() >= DEDUP_CAPACITY
600            && self
601                .clearing
602                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
603                .is_ok()
604        {
605            set.clear();
606            self.clearing.store(false, Ordering::Release);
607        }
608    }
609}
610
611/// Outcome of a single dispatch call.
612#[derive(Debug, Clone, Copy, PartialEq, Eq)]
613pub enum DispatchOutcome {
614    /// The report was for a tracked order. Typed events have been emitted
615    /// (or intentionally skipped, e.g. dedup hit). The caller must not
616    /// forward the report as a fallback.
617    Tracked,
618    /// The report is for an external / untracked order. The caller should
619    /// forward the report via [`ExecutionEventEmitter::send_order_status_report`]
620    /// or [`ExecutionEventEmitter::send_fill_report`] so the engine can
621    /// reconcile.
622    External,
623    /// The report was recognised as stale (e.g. cancel leg of a
624    /// cancel-replace modify, or replay after terminal state). The caller
625    /// must drop it without forwarding.
626    Skip,
627}
628
629/// Dispatches an [`OrderStatusReport`] using the two-tier routing contract.
630///
631/// Returns [`DispatchOutcome::Tracked`] when the report maps to a tracked
632/// order (typed events have been emitted or dedup hit), [`External`] when
633/// the caller should forward the report as an untracked fallback, or
634/// [`Skip`] when the report is a stale / race leg that must be dropped.
635///
636/// [`External`]: DispatchOutcome::External
637/// [`Skip`]: DispatchOutcome::Skip
638pub fn dispatch_order_event(
639    report: &OrderStatusReport,
640    state: &WsDispatchState,
641    emitter: &ExecutionEventEmitter,
642    ts_init: UnixNanos,
643) -> DispatchOutcome {
644    let Some(client_order_id) = report.client_order_id else {
645        return DispatchOutcome::External;
646    };
647
648    if state.filled_orders.contains(&client_order_id) {
649        log::debug!(
650            "Skipping stale report for filled order: cid={client_order_id}, status={:?}",
651            report.order_status,
652        );
653        return DispatchOutcome::Skip;
654    }
655
656    let client_order_id_str = client_order_id.as_str();
657    if client_order_id_str.starts_with("0x")
658        && state.terminal_cloid_seen(&Ustr::from(client_order_id_str))
659    {
660        log::debug!(
661            "Skipping stale terminal report for raw cloid: cid={client_order_id}, status={:?}",
662            report.order_status,
663        );
664        return DispatchOutcome::Skip;
665    }
666
667    let Some(context) = state.lookup_context(&client_order_id) else {
668        return DispatchOutcome::External;
669    };
670
671    match report.order_status {
672        OrderStatus::Accepted => {
673            handle_accepted(report, client_order_id, &context, state, emitter, ts_init)
674        }
675        OrderStatus::Triggered => {
676            handle_triggered(report, client_order_id, &context, state, emitter, ts_init)
677        }
678        OrderStatus::Canceled => {
679            handle_canceled(report, client_order_id, &context, state, emitter, ts_init)
680        }
681        OrderStatus::Expired => {
682            handle_expired(report, client_order_id, &context, state, emitter, ts_init)
683        }
684        OrderStatus::Rejected => {
685            handle_rejected(report, client_order_id, &context, state, emitter, ts_init)
686        }
687        OrderStatus::Filled => handle_filled_marker(client_order_id, state),
688        OrderStatus::PartiallyFilled => {
689            // Fills come via `FillReport`; nothing to emit from the status path.
690            DispatchOutcome::Tracked
691        }
692        OrderStatus::PendingUpdate
693        | OrderStatus::PendingCancel
694        | OrderStatus::Submitted
695        | OrderStatus::Initialized
696        | OrderStatus::Denied
697        | OrderStatus::Released
698        | OrderStatus::Emulated
699        | OrderStatus::Voided => DispatchOutcome::Tracked,
700    }
701}
702
703/// Dispatches a [`FillReport`] using the two-tier routing contract.
704///
705/// Returns [`DispatchOutcome::Tracked`] when the fill has been emitted as
706/// an `OrderFilled` event (or skipped via trade dedup), [`External`] when
707/// the caller should forward the fill via
708/// [`ExecutionEventEmitter::send_fill_report`], or [`Skip`] when the fill
709/// is a replay for an already-terminal order and must be dropped.
710///
711/// [`External`]: DispatchOutcome::External
712/// [`Skip`]: DispatchOutcome::Skip
713pub fn dispatch_order_fill(
714    report: &FillReport,
715    state: &WsDispatchState,
716    emitter: &ExecutionEventEmitter,
717    ts_init: UnixNanos,
718) -> DispatchOutcome {
719    let Some(client_order_id) = report.client_order_id else {
720        return DispatchOutcome::External;
721    };
722
723    if state.filled_orders.contains(&client_order_id) {
724        log::debug!(
725            "Skipping stale fill for filled order: cid={client_order_id}, trade_id={}",
726            report.trade_id,
727        );
728        return DispatchOutcome::Skip;
729    }
730
731    let Some(mut context) = state.lookup_context(&client_order_id) else {
732        return DispatchOutcome::External;
733    };
734
735    // Set when a fill promotes, so the corrective-reduce runs after the fill applies
736    let mut promoted_corrective: Option<(Quantity, HyperliquidExchangePlaceOrderRequest)> = None;
737
738    // Promote the binding from the fill so a dropped replacement ACCEPTED cannot
739    // strand it (see module docs).
740    if state.has_pending_modify(&client_order_id)
741        && let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
742        && report.venue_order_id != cached_voi
743    {
744        let target = state.pending_modify_target_qty(&client_order_id);
745        let sent_request = state.modify_request(&client_order_id);
746        // Prefer the modify target price over the stale cached context price
747        let price = sent_request
748            .as_ref()
749            .zip(context.price)
750            .and_then(|(r, cached)| Price::from_decimal_dp(r.price, cached.precision).ok())
751            .or(context.price);
752        let Some(price) = price else {
753            log::warn!(
754                "Cannot promote cancel-replace for {client_order_id} from fill: no target \
755                 or cached price; buffering until the replacement ACCEPTED arrives",
756            );
757            state.buffer_fill(client_order_id, report.clone());
758            return DispatchOutcome::Tracked;
759        };
760        let updated_quantity = target.unwrap_or(context.quantity);
761        promote_cancel_replace(
762            client_order_id,
763            &context,
764            state,
765            emitter,
766            report.venue_order_id,
767            report.account_id,
768            price,
769            updated_quantity,
770            None,
771            report.ts_event,
772            ts_init,
773        );
774        // Re-read the context advanced by the promotion (quantity and price)
775        if let Some(updated) = state.lookup_context(&client_order_id) {
776            context = updated;
777        }
778
779        if let (Some(target), Some(sent_request)) = (target, sent_request) {
780            promoted_corrective = Some((target, sent_request));
781        }
782    }
783
784    if state.check_and_insert_trade(report.trade_id) {
785        log::debug!(
786            "Skipping duplicate fill for {client_order_id}: trade_id={}",
787            report.trade_id
788        );
789        return DispatchOutcome::Tracked;
790    }
791
792    let previous = state
793        .previous_filled_qty(&client_order_id)
794        .unwrap_or_else(|| Quantity::zero(report.last_qty.precision));
795    let cumulative = previous + report.last_qty;
796
797    let is_terminal_fill = cumulative >= context.quantity;
798    if is_terminal_fill && !claim_terminal_order(client_order_id, state, OrderStatus::Filled) {
799        return DispatchOutcome::Skip;
800    }
801
802    ensure_accepted_emitted(
803        client_order_id,
804        report.venue_order_id,
805        report.account_id,
806        &context,
807        state,
808        emitter,
809        report.ts_event,
810        ts_init,
811    );
812
813    let filled = OrderFilled::new(
814        emitter.trader_id(),
815        context.identity.strategy_id,
816        context.identity.instrument_id,
817        client_order_id,
818        report.venue_order_id,
819        report.account_id,
820        report.trade_id,
821        context.identity.order_side,
822        context.identity.order_type,
823        report.last_qty,
824        report.last_px,
825        report.commission.currency,
826        report.liquidity_side,
827        UUID4::new(),
828        report.ts_event,
829        ts_init,
830        false,
831        report.venue_position_id,
832        Some(report.commission),
833        None,
834    );
835    emitter.send_order_event(OrderEventAny::Filled(filled));
836
837    state.record_filled_qty(client_order_id, cumulative);
838
839    // Cumulative now includes this fill, so the reduce sizes against the true remaining
840    if let Some((target, sent_request)) = promoted_corrective {
841        maybe_queue_corrective_reduce(
842            state,
843            client_order_id,
844            report.venue_order_id,
845            target,
846            sent_request,
847        );
848    }
849
850    if is_terminal_fill {
851        state.cleanup_terminal(&client_order_id);
852    }
853
854    DispatchOutcome::Tracked
855}
856
857fn handle_accepted(
858    report: &OrderStatusReport,
859    client_order_id: ClientOrderId,
860    context: &OrderContext,
861    state: &WsDispatchState,
862    emitter: &ExecutionEventEmitter,
863    ts_init: UnixNanos,
864) -> DispatchOutcome {
865    let venue_order_id = report.venue_order_id;
866    let ts_event = report.ts_last;
867    let account_id = report.account_id;
868
869    // Cancel-replace detection: if an earlier ACCEPTED cached a different
870    // venue_order_id under the same client_order_id, this ACCEPTED is the
871    // replacement leg of a Hyperliquid modify and must be promoted to
872    // OrderUpdated. See GH-3827.
873    if let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
874        && cached_voi != venue_order_id
875    {
876        let price = report.price.or(context.price);
877        let Some(price) = price else {
878            log::warn!(
879                "Cannot emit OrderUpdated for cancel-replace {client_order_id}: \
880                 no price on report and no cached price on context",
881            );
882            return DispatchOutcome::Skip;
883        };
884
885        // Prefer user target over venue's remaining-only `report.quantity`;
886        // fall back when no marker (external modify).
887        let target_total_qty = state.pending_modify_target_qty(&client_order_id);
888        let updated_quantity = target_total_qty.unwrap_or(report.quantity);
889        let sent_request = state.modify_request(&client_order_id);
890
891        promote_cancel_replace(
892            client_order_id,
893            context,
894            state,
895            emitter,
896            venue_order_id,
897            account_id,
898            price,
899            updated_quantity,
900            report.trigger_price,
901            ts_event,
902            ts_init,
903        );
904
905        if let (Some(target), Some(sent_request)) = (target_total_qty, sent_request) {
906            maybe_queue_corrective_reduce(
907                state,
908                client_order_id,
909                venue_order_id,
910                target,
911                sent_request,
912            );
913        }
914
915        return DispatchOutcome::Tracked;
916    }
917
918    if state.emitted_accepted.contains(&client_order_id) {
919        // Repeat ACCEPTED for an already-accepted order. Nothing to emit;
920        // refresh the cached price so a subsequent cancel-replace without a
921        // report price can still recover an accurate value.
922        state.update_context_price(&client_order_id, report.price);
923        return DispatchOutcome::Tracked;
924    }
925
926    state.insert_accepted(client_order_id);
927    state.record_venue_order_id(client_order_id, venue_order_id);
928    state.update_context_price(&client_order_id, report.price);
929
930    let accepted = OrderAccepted::new(
931        emitter.trader_id(),
932        context.identity.strategy_id,
933        context.identity.instrument_id,
934        client_order_id,
935        venue_order_id,
936        account_id,
937        UUID4::new(),
938        ts_event,
939        ts_init,
940        false,
941    );
942    emitter.send_order_event(OrderEventAny::Accepted(accepted));
943    DispatchOutcome::Tracked
944}
945
946// Shared by the ACCEPTED branch and the fill path (dropped-ACCEPTED recovery) so the
947// cancel-replace binding is recovered from whichever arrives first. See GH-3827, GH-3972.
948#[allow(
949    clippy::too_many_arguments,
950    reason = "promotion needs the full OrderUpdated field set, sourced from two report shapes"
951)]
952fn promote_cancel_replace(
953    client_order_id: ClientOrderId,
954    context: &OrderContext,
955    state: &WsDispatchState,
956    emitter: &ExecutionEventEmitter,
957    venue_order_id: VenueOrderId,
958    account_id: AccountId,
959    price: Price,
960    quantity: Quantity,
961    trigger_price: Option<Price>,
962    ts_event: UnixNanos,
963    ts_init: UnixNanos,
964) {
965    state.record_venue_order_id(client_order_id, venue_order_id);
966    state.update_context_quantity(&client_order_id, quantity);
967    state.update_context_price(&client_order_id, Some(price));
968    // Claim the front intent; the next queued modify advances to this replacement
969    state.claim_front_modify(&client_order_id, venue_order_id);
970
971    let updated = OrderUpdated::new(
972        emitter.trader_id(),
973        context.identity.strategy_id,
974        context.identity.instrument_id,
975        client_order_id,
976        quantity,
977        UUID4::new(),
978        ts_event,
979        ts_init,
980        false,
981        Some(venue_order_id),
982        Some(account_id),
983        Some(price),
984        trigger_price,
985        None,
986        false,
987    );
988    emitter.send_order_event(OrderEventAny::Updated(updated));
989
990    // Drain fills buffered before the binding advanced. Bypasses
991    // `handle_execution_report`; FIFO-bounded caches make any residue benign.
992    let buffered = state.drain_buffered_fills(&client_order_id);
993    for fill in buffered {
994        dispatch_order_fill(&fill, state, emitter, ts_init);
995    }
996}
997
998/// Promotes a cancel-replace replacement surfaced by a query during an in-flight modify.
999///
1000/// When the query returns the replacement leg (`Accepted`, `venue_order_id` diverging from the
1001/// cached one, modify tracked), emits the `OrderUpdated` that rebinds the order, so a dropped
1002/// replacement `Accepted` with no fill cannot strand the binding on the canceled leg. Returns
1003/// `true` when promoted; the caller still forwards the report so the engine confirms the order.
1004pub fn promote_replacement_from_query(
1005    report: &OrderStatusReport,
1006    state: &WsDispatchState,
1007    emitter: &ExecutionEventEmitter,
1008    ts_init: UnixNanos,
1009) -> bool {
1010    if report.order_status != OrderStatus::Accepted {
1011        return false;
1012    }
1013
1014    let Some(client_order_id) = report.client_order_id else {
1015        return false;
1016    };
1017
1018    if !state.has_pending_modify(&client_order_id) {
1019        return false;
1020    }
1021
1022    let Some(cached_voi) = state.cached_venue_order_id(&client_order_id) else {
1023        return false;
1024    };
1025
1026    if report.venue_order_id == cached_voi {
1027        return false;
1028    }
1029
1030    let Some(context) = state.lookup_context(&client_order_id) else {
1031        return false;
1032    };
1033
1034    let Some(price) = report.price.or(context.price) else {
1035        log::warn!(
1036            "Cannot promote cancel-replace from query for {client_order_id}: \
1037             no price on report and no cached price on context",
1038        );
1039        return false;
1040    };
1041
1042    // Prefer the user target over the venue's remaining-only `report.quantity`
1043    let updated_quantity = state
1044        .pending_modify_target_qty(&client_order_id)
1045        .unwrap_or(report.quantity);
1046
1047    promote_cancel_replace(
1048        client_order_id,
1049        &context,
1050        state,
1051        emitter,
1052        report.venue_order_id,
1053        report.account_id,
1054        price,
1055        updated_quantity,
1056        report.trigger_price,
1057        report.ts_last,
1058        ts_init,
1059    );
1060
1061    log::debug!("Promoted cancel-replace replacement for {client_order_id} from query");
1062
1063    true
1064}
1065
1066// Queue a corrective reduce when a fill that raced the modify left the replacement
1067// oversized. Reached from both promotion paths; the engine overfill guard backstops.
1068fn maybe_queue_corrective_reduce(
1069    state: &WsDispatchState,
1070    client_order_id: ClientOrderId,
1071    venue_order_id: VenueOrderId,
1072    target: Quantity,
1073    sent_request: HyperliquidExchangePlaceOrderRequest,
1074) {
1075    let Ok(new_oid) = venue_order_id.as_str().parse::<u64>() else {
1076        return;
1077    };
1078
1079    let filled = state
1080        .previous_filled_qty(&client_order_id)
1081        .unwrap_or_else(|| Quantity::zero(target.precision));
1082    if filled >= target {
1083        return;
1084    }
1085
1086    let remaining = (target - filled).as_decimal().normalize();
1087
1088    let sent_size = sent_request.size;
1089    if sent_size > remaining {
1090        let mut corrective = sent_request;
1091        corrective.size = remaining;
1092
1093        state.mark_pending_modify(client_order_id, venue_order_id, target);
1094        state.stash_modify_request(client_order_id, corrective.clone());
1095        state.queue_corrective(client_order_id, new_oid, corrective);
1096
1097        log::warn!(
1098            "Cancel-replace left {client_order_id} oversized on {venue_order_id} \
1099             (sent {sent_size}, remaining {remaining}); queuing corrective reduce",
1100        );
1101    }
1102}
1103
1104fn handle_triggered(
1105    report: &OrderStatusReport,
1106    client_order_id: ClientOrderId,
1107    context: &OrderContext,
1108    state: &WsDispatchState,
1109    emitter: &ExecutionEventEmitter,
1110    ts_init: UnixNanos,
1111) -> DispatchOutcome {
1112    if !matches!(
1113        context.identity.order_type,
1114        OrderType::StopLimit | OrderType::TrailingStopLimit | OrderType::LimitIfTouched
1115    ) {
1116        log::debug!(
1117            "Ignoring TRIGGERED status for non-triggerable order type {:?}: {client_order_id}",
1118            context.identity.order_type,
1119        );
1120        return DispatchOutcome::Tracked;
1121    }
1122
1123    ensure_accepted_emitted(
1124        client_order_id,
1125        report.venue_order_id,
1126        report.account_id,
1127        context,
1128        state,
1129        emitter,
1130        report.ts_last,
1131        ts_init,
1132    );
1133
1134    let triggered = OrderTriggered::new(
1135        emitter.trader_id(),
1136        context.identity.strategy_id,
1137        context.identity.instrument_id,
1138        client_order_id,
1139        UUID4::new(),
1140        report.ts_last,
1141        ts_init,
1142        false,
1143        Some(report.venue_order_id),
1144        Some(report.account_id),
1145    );
1146    emitter.send_order_event(OrderEventAny::Triggered(triggered));
1147    DispatchOutcome::Tracked
1148}
1149
1150fn handle_canceled(
1151    report: &OrderStatusReport,
1152    client_order_id: ClientOrderId,
1153    context: &OrderContext,
1154    state: &WsDispatchState,
1155    emitter: &ExecutionEventEmitter,
1156    ts_init: UnixNanos,
1157) -> DispatchOutcome {
1158    let venue_order_id = report.venue_order_id;
1159
1160    // Stale cancel suppression: if the cached venue_order_id has already
1161    // been advanced by a cancel-replace promotion, this CANCELED refers to
1162    // the old leg and has already been handled as OrderUpdated. See GH-3827.
1163    if let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
1164        && cached_voi != venue_order_id
1165    {
1166        log::debug!(
1167            "Skipping stale CANCELED for {venue_order_id} (cached {cached_voi}) on {client_order_id}",
1168        );
1169        return DispatchOutcome::Skip;
1170    }
1171
1172    // Cancel-before-accept race: an in-flight modify may deliver
1173    // CANCELED(old_voi) before the replacement ACCEPTED(new_voi). Any queued
1174    // intent whose old leg matches (marked before the HTTP call, cleared on
1175    // failure) suppresses that cancel so the later ACCEPTED routes through
1176    // OrderUpdated. See GH-3827.
1177    if state.pending_modify_contains_old(&client_order_id, venue_order_id) {
1178        log::debug!(
1179            "Skipping cancel-before-accept leg for {client_order_id}: venue_order_id={venue_order_id}",
1180        );
1181        return DispatchOutcome::Skip;
1182    }
1183
1184    if !claim_terminal_order(client_order_id, state, report.order_status) {
1185        return DispatchOutcome::Skip;
1186    }
1187
1188    ensure_accepted_emitted(
1189        client_order_id,
1190        venue_order_id,
1191        report.account_id,
1192        context,
1193        state,
1194        emitter,
1195        report.ts_last,
1196        ts_init,
1197    );
1198
1199    let canceled = OrderCanceled::new(
1200        emitter.trader_id(),
1201        context.identity.strategy_id,
1202        context.identity.instrument_id,
1203        client_order_id,
1204        UUID4::new(),
1205        report.ts_last,
1206        ts_init,
1207        false,
1208        Some(venue_order_id),
1209        Some(report.account_id),
1210    );
1211    emitter.send_order_event(OrderEventAny::Canceled(canceled));
1212
1213    state.cleanup_terminal(&client_order_id);
1214    DispatchOutcome::Tracked
1215}
1216
1217fn handle_expired(
1218    report: &OrderStatusReport,
1219    client_order_id: ClientOrderId,
1220    context: &OrderContext,
1221    state: &WsDispatchState,
1222    emitter: &ExecutionEventEmitter,
1223    ts_init: UnixNanos,
1224) -> DispatchOutcome {
1225    if !claim_terminal_order(client_order_id, state, report.order_status) {
1226        return DispatchOutcome::Skip;
1227    }
1228
1229    ensure_accepted_emitted(
1230        client_order_id,
1231        report.venue_order_id,
1232        report.account_id,
1233        context,
1234        state,
1235        emitter,
1236        report.ts_last,
1237        ts_init,
1238    );
1239
1240    let expired = OrderExpired::new(
1241        emitter.trader_id(),
1242        context.identity.strategy_id,
1243        context.identity.instrument_id,
1244        client_order_id,
1245        UUID4::new(),
1246        report.ts_last,
1247        ts_init,
1248        false,
1249        Some(report.venue_order_id),
1250        Some(report.account_id),
1251    );
1252    emitter.send_order_event(OrderEventAny::Expired(expired));
1253    state.cleanup_terminal(&client_order_id);
1254    DispatchOutcome::Tracked
1255}
1256
1257fn handle_rejected(
1258    report: &OrderStatusReport,
1259    client_order_id: ClientOrderId,
1260    context: &OrderContext,
1261    state: &WsDispatchState,
1262    emitter: &ExecutionEventEmitter,
1263    ts_init: UnixNanos,
1264) -> DispatchOutcome {
1265    if state.submission_pending(&client_order_id) {
1266        state.buffer_submission_rejection(client_order_id, report.clone());
1267        return DispatchOutcome::Skip;
1268    }
1269
1270    if !claim_terminal_order(client_order_id, state, report.order_status) {
1271        return DispatchOutcome::Skip;
1272    }
1273
1274    let reason = report
1275        .cancel_reason
1276        .clone()
1277        .unwrap_or_else(|| "Order rejected by exchange".to_string());
1278    let rejected = OrderRejected::new(
1279        emitter.trader_id(),
1280        context.identity.strategy_id,
1281        context.identity.instrument_id,
1282        client_order_id,
1283        report.account_id,
1284        Ustr::from(&reason),
1285        UUID4::new(),
1286        report.ts_last,
1287        ts_init,
1288        false,
1289        report.post_only && reason.contains(HYPERLIQUID_POST_ONLY_WOULD_MATCH),
1290    );
1291    emitter.send_order_event(OrderEventAny::Rejected(rejected));
1292    state.cleanup_terminal(&client_order_id);
1293    DispatchOutcome::Tracked
1294}
1295
1296fn claim_terminal_order(
1297    client_order_id: ClientOrderId,
1298    state: &WsDispatchState,
1299    status: OrderStatus,
1300) -> bool {
1301    let claimed = state.insert_filled(client_order_id);
1302    if !claimed {
1303        log::debug!("Skipping duplicate terminal event for {client_order_id}: status={status:?}",);
1304    }
1305
1306    claimed
1307}
1308
1309fn handle_filled_marker(
1310    _client_order_id: ClientOrderId,
1311    _state: &WsDispatchState,
1312) -> DispatchOutcome {
1313    // A status-only `FILLED` marker does not carry fill data; the actual
1314    // `OrderFilled` is emitted from `dispatch_order_fill` when the matching
1315    // trade arrives. Do *not* set `filled_orders` here, otherwise the
1316    // follow-up fill would be classified as a stale replay and dropped
1317    // before the terminal `OrderFilled` event can be emitted. The fill
1318    // path installs the marker itself once the cumulative fill quantity
1319    // matches the tracked order quantity.
1320    DispatchOutcome::Tracked
1321}
1322
1323/// Synthesizes and emits an `OrderAccepted` event when one has not yet been
1324/// emitted for the given order.
1325///
1326/// Used before emitting non-Accepted events so strategies always observe the
1327/// canonical `Submitted -> Accepted -> ...` lifecycle even when the venue
1328/// compresses the placement and follow-up event into a single message (fast
1329/// fills).
1330#[allow(clippy::too_many_arguments)]
1331fn ensure_accepted_emitted(
1332    client_order_id: ClientOrderId,
1333    venue_order_id: VenueOrderId,
1334    account_id: AccountId,
1335    context: &OrderContext,
1336    state: &WsDispatchState,
1337    emitter: &ExecutionEventEmitter,
1338    ts_event: UnixNanos,
1339    ts_init: UnixNanos,
1340) {
1341    if state.emitted_accepted.contains(&client_order_id) {
1342        return;
1343    }
1344    state.insert_accepted(client_order_id);
1345    state.record_venue_order_id(client_order_id, venue_order_id);
1346
1347    let accepted = OrderAccepted::new(
1348        emitter.trader_id(),
1349        context.identity.strategy_id,
1350        context.identity.instrument_id,
1351        client_order_id,
1352        venue_order_id,
1353        account_id,
1354        UUID4::new(),
1355        ts_event,
1356        ts_init,
1357        false,
1358    );
1359    emitter.send_order_event(OrderEventAny::Accepted(accepted));
1360}
1361
1362#[cfg(test)]
1363mod tests {
1364    use nautilus_live::execution::context::OrderIdentity;
1365    use nautilus_model::{
1366        enums::{OrderSide, TimeInForce},
1367        identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId},
1368    };
1369    use rstest::rstest;
1370    use rust_decimal::Decimal;
1371
1372    use super::*;
1373    use crate::http::models::{
1374        HyperliquidExchangeLimitParams, HyperliquidExchangeOrderKind, HyperliquidExchangeTif,
1375    };
1376
1377    fn make_context(client_order_id: ClientOrderId) -> OrderContext {
1378        OrderContext {
1379            identity: OrderIdentity {
1380                client_order_id,
1381                strategy_id: StrategyId::from("S-001"),
1382                instrument_id: InstrumentId::from("BTC-USD-PERP.HYPERLIQUID"),
1383                order_side: OrderSide::Buy,
1384                order_type: OrderType::Limit,
1385            },
1386            quantity: Quantity::from("0.0001"),
1387            price: None,
1388            trigger_price: None,
1389            trigger_type: None,
1390            time_in_force: TimeInForce::Gtc,
1391            is_post_only: false,
1392            is_reduce_only: false,
1393            is_quote_quantity: false,
1394        }
1395    }
1396
1397    #[rstest]
1398    fn test_register_and_lookup_context() {
1399        let state = WsDispatchState::new();
1400        let cid = ClientOrderId::new("O-001");
1401        state.register_context(make_context(cid));
1402
1403        assert_eq!(state.lookup_context(&cid), Some(make_context(cid)));
1404    }
1405
1406    #[rstest]
1407    fn test_lookup_context_missing_returns_none() {
1408        let state = WsDispatchState::new();
1409        let cid = ClientOrderId::new("not-tracked");
1410        assert!(state.lookup_context(&cid).is_none());
1411    }
1412
1413    #[rstest]
1414    fn test_update_context_refreshes_price_and_quantity() {
1415        let state = WsDispatchState::new();
1416        let cid = ClientOrderId::new("O-003");
1417        state.register_context(make_context(cid));
1418
1419        state.update_context_price(&cid, Some(Price::from("56731.5")));
1420        state.update_context_quantity(&cid, Quantity::from("0.0002"));
1421
1422        assert_eq!(
1423            state.lookup_context(&cid),
1424            Some(OrderContext {
1425                price: Some(Price::from("56731.5")),
1426                quantity: Quantity::from("0.0002"),
1427                ..make_context(cid)
1428            }),
1429        );
1430    }
1431
1432    #[rstest]
1433    fn test_update_context_price_without_price_keeps_cached_price() {
1434        let state = WsDispatchState::new();
1435        let cid = ClientOrderId::new("O-004");
1436        let cached = OrderContext {
1437            price: Some(Price::from("56730.0")),
1438            ..make_context(cid)
1439        };
1440        state.register_context(cached);
1441
1442        state.update_context_price(&cid, None);
1443
1444        assert_eq!(state.lookup_context(&cid), Some(cached));
1445    }
1446
1447    #[rstest]
1448    fn test_insert_accepted_dedup() {
1449        let state = WsDispatchState::new();
1450        let cid = ClientOrderId::new("O-002");
1451        assert!(!state.emitted_accepted.contains(&cid));
1452        state.insert_accepted(cid);
1453        assert!(state.emitted_accepted.contains(&cid));
1454        state.insert_accepted(cid);
1455        assert!(state.emitted_accepted.contains(&cid));
1456    }
1457
1458    #[rstest]
1459    fn test_check_and_insert_trade_detects_duplicates() {
1460        let state = WsDispatchState::new();
1461        let trade = TradeId::new("trade-1");
1462        assert!(!state.check_and_insert_trade(trade));
1463        assert!(state.check_and_insert_trade(trade));
1464    }
1465
1466    #[rstest]
1467    fn test_pending_modify_roundtrip() {
1468        let state = WsDispatchState::new();
1469        let cid = ClientOrderId::new("O-010");
1470        let voi = VenueOrderId::new("v-1");
1471        let target_qty = Quantity::from("0.0001");
1472
1473        assert!(state.pending_modify(&cid).is_none());
1474        assert!(state.pending_modify_target_qty(&cid).is_none());
1475        state.mark_pending_modify(cid, voi, target_qty);
1476        assert_eq!(state.pending_modify(&cid), Some(voi));
1477        assert_eq!(state.pending_modify_target_qty(&cid), Some(target_qty));
1478        state.clear_pending_modify(&cid);
1479        assert!(state.pending_modify(&cid).is_none());
1480        assert!(state.pending_modify_target_qty(&cid).is_none());
1481    }
1482
1483    #[rstest]
1484    fn test_cleanup_terminal_preserves_filled_marker() {
1485        let state = WsDispatchState::new();
1486        let cid = ClientOrderId::new("O-020");
1487        state.register_context(make_context(cid));
1488        state.insert_accepted(cid);
1489        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.0001"));
1490        state.insert_filled(cid);
1491        state.cleanup_terminal(&cid);
1492
1493        assert!(state.lookup_context(&cid).is_none());
1494        assert!(!state.emitted_accepted.contains(&cid));
1495        assert!(state.pending_modify(&cid).is_none());
1496        assert!(state.pending_modify_target_qty(&cid).is_none());
1497        // `filled_orders` outlives `cleanup_terminal` so replays stay suppressed.
1498        assert!(state.filled_orders.contains(&cid));
1499    }
1500
1501    #[rstest]
1502    fn test_cleanup_terminal_clears_corrective_state() {
1503        let state = WsDispatchState::new();
1504        let cid = ClientOrderId::new("O-021");
1505        let request = sample_request(Decimal::from(1));
1506        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("1"));
1507        state.stash_modify_request(cid, request.clone());
1508        state.queue_corrective(cid, 1, request);
1509        assert!(state.modify_request(&cid).is_some());
1510
1511        state.cleanup_terminal(&cid);
1512
1513        assert!(state.modify_request(&cid).is_none());
1514        assert!(state.take_corrective(&cid).is_none());
1515        assert!(state.pending_modify(&cid).is_none());
1516    }
1517
1518    fn sample_request(size: Decimal) -> HyperliquidExchangePlaceOrderRequest {
1519        HyperliquidExchangePlaceOrderRequest {
1520            asset: 0,
1521            is_buy: true,
1522            price: "100".parse::<Decimal>().unwrap(),
1523            size,
1524            reduce_only: false,
1525            kind: HyperliquidExchangeOrderKind::Limit {
1526                limit: HyperliquidExchangeLimitParams {
1527                    tif: HyperliquidExchangeTif::Gtc,
1528                },
1529            },
1530            cloid: None,
1531        }
1532    }
1533
1534    #[rstest]
1535    fn test_modify_chain_keeps_both_intents_on_rapid_modifies() {
1536        let state = WsDispatchState::new();
1537        let cid = ClientOrderId::new("O-100");
1538        let g0 =
1539            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
1540        let g1 =
1541            state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00030"));
1542
1543        assert_ne!(g0, g1);
1544        assert!(state.has_pending_modify(&cid));
1545        // Front is the oldest intent
1546        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
1547        assert_eq!(
1548            state.pending_modify_target_qty(&cid),
1549            Some(Quantity::from("0.00020")),
1550        );
1551        // Both queued old legs suppress their cancel-before-accept
1552        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
1553        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
1554    }
1555
1556    #[rstest]
1557    fn test_clear_modify_generation_preserves_newer_intent() {
1558        let state = WsDispatchState::new();
1559        let cid = ClientOrderId::new("O-101");
1560        // Two rapid modifies queued before either acked, both against the live
1561        // leg v-0.
1562        let g0 =
1563            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
1564        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));
1565
1566        // Failure of the first modify clears only its generation; the second
1567        // stays, still targeting the live leg v-0.
1568        state.clear_modify_generation(&cid, g0);
1569
1570        assert!(state.has_pending_modify(&cid));
1571        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
1572        assert_eq!(
1573            state.pending_modify_target_qty(&cid),
1574            Some(Quantity::from("0.00030")),
1575        );
1576        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
1577    }
1578
1579    #[rstest]
1580    fn test_claim_front_modify_advances_next_old_id() {
1581        let state = WsDispatchState::new();
1582        let cid = ClientOrderId::new("O-102");
1583        // Both queued against the same stale old leg (M2 fired before M1 acked)
1584        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
1585        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));
1586
1587        // Promoting the first replacement claims the front and advances the
1588        // next intent's old leg to the replacement id.
1589        let claimed = state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
1590        assert_eq!(
1591            claimed.map(|i| i.target_qty),
1592            Some(Quantity::from("0.00020"))
1593        );
1594
1595        assert!(state.has_pending_modify(&cid));
1596        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
1597        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
1598        // The stale leg no longer matches once advanced
1599        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
1600
1601        // Claiming the last intent empties the chain
1602        let claimed2 = state.claim_front_modify(&cid, VenueOrderId::new("v-2"));
1603        assert_eq!(
1604            claimed2.map(|i| i.target_qty),
1605            Some(Quantity::from("0.00030"))
1606        );
1607        assert!(!state.has_pending_modify(&cid));
1608        assert!(state.pending_modify(&cid).is_none());
1609    }
1610
1611    #[rstest]
1612    fn test_modify_chain_caps_and_evicts_oldest() {
1613        let state = WsDispatchState::new();
1614        let cid = ClientOrderId::new("O-106");
1615        // Queue one past the cap with no promotion or clear to drain them
1616        for i in 0..=MAX_PENDING_MODIFY_INTENTS {
1617            let voi = format!("v-{i}");
1618            state.mark_pending_modify(cid, VenueOrderId::new(&voi), Quantity::from("0.00020"));
1619        }
1620
1621        // The oldest intent was evicted; the newest remains and the front
1622        // advanced to the second-oldest.
1623        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
1624        let newest = format!("v-{MAX_PENDING_MODIFY_INTENTS}");
1625        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new(&newest)));
1626        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
1627    }
1628
1629    #[rstest]
1630    fn test_clear_front_modify_reparents_next_old() {
1631        let state = WsDispatchState::new();
1632        let cid = ClientOrderId::new("O-104");
1633        // Three rapid modifies where the first already promoted to v-1
1634        // (advancing the front to old=v-1); the third still holds stale v-0.
1635        let g_front =
1636            state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00020"));
1637        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));
1638
1639        // The front is rejected; the next intent must inherit the live leg
1640        // (v-1), not keep stale v-0, or CANCELED(v-1) would surface as a real
1641        // cancel.
1642        state.clear_modify_generation(&cid, g_front);
1643
1644        assert!(state.has_pending_modify(&cid));
1645        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
1646        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
1647        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
1648    }
1649
1650    #[rstest]
1651    fn test_clear_non_front_modify_leaves_front_old() {
1652        let state = WsDispatchState::new();
1653        let cid = ClientOrderId::new("O-105");
1654        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
1655        let g_back =
1656            state.mark_pending_modify(cid, VenueOrderId::new("v-9"), Quantity::from("0.00030"));
1657
1658        // Removing a non-front intent must not disturb the front's old leg
1659        state.clear_modify_generation(&cid, g_back);
1660
1661        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
1662        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-9")));
1663    }
1664
1665    #[rstest]
1666    fn test_stash_modify_request_targets_latest_intent() {
1667        let state = WsDispatchState::new();
1668        let cid = ClientOrderId::new("O-103");
1669        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
1670        state.stash_modify_request(cid, sample_request(Decimal::from(1)));
1671        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00030"));
1672        state.stash_modify_request(cid, sample_request(Decimal::from(2)));
1673
1674        // Front intent keeps its own request
1675        assert_eq!(
1676            state.modify_request(&cid).map(|r| r.size),
1677            Some(Decimal::from(1)),
1678        );
1679        // After claiming the front, the next intent's request surfaces
1680        state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
1681        assert_eq!(
1682            state.modify_request(&cid).map(|r| r.size),
1683            Some(Decimal::from(2)),
1684        );
1685    }
1686}
nautilus_hyperliquid/websocket/dispatch.rs

nautilus_hyperliquid/websocket/
dispatch.rs
