Skip to main content

nautilus_betfair/stream/
ocm.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 OCM stream handler state.
17
18use std::collections::VecDeque;
19
20use ahash::{AHashMap, AHashSet};
21use nautilus_core::UUID4;
22use nautilus_model::{
23    identifiers::{ClientOrderId, StrategyId, VenueOrderId},
24    types::Quantity,
25};
26use rust_decimal::Decimal;
27
28use crate::{
29    common::{
30        parse::{make_customer_order_ref, make_customer_order_ref_legacy},
31        types::{BetId, CustomerOrderRef, OrderSyncEntry},
32    },
33    stream::{messages::UnmatchedOrder, parse::FillTracker},
34};
35
36#[derive(Clone, Debug)]
37pub(crate) struct PendingReplaceState {
38    id: UUID4,
39    pub(crate) total_quantity: Option<Quantity>,
40    awaiting_reconciliation: bool,
41}
42
43#[derive(Debug, Clone, Copy)]
44struct PendingReductionState {
45    client_order_id: ClientOrderId,
46    original_quantity: Quantity,
47    requested_quantity: Quantity,
48    confirmed_quantity: Option<Quantity>,
49}
50
51impl PendingReductionState {
52    fn is_unconfirmed_for(&self, client_order_id: &ClientOrderId) -> bool {
53        self.client_order_id == *client_order_id && self.confirmed_quantity.is_none()
54    }
55
56    fn can_confirm(&self, client_order_id: &ClientOrderId, active_quantity: Quantity) -> bool {
57        self.is_unconfirmed_for(client_order_id)
58            && active_quantity >= self.requested_quantity
59            && active_quantity < self.original_quantity
60    }
61}
62
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64pub(crate) enum CustomerOrderRefResolution {
65    Unique(ClientOrderId),
66    Ambiguous,
67}
68
69impl CustomerOrderRefResolution {
70    pub(crate) fn client_order_id(self) -> Option<ClientOrderId> {
71        match self {
72            Self::Unique(client_order_id) => Some(client_order_id),
73            Self::Ambiguous => None,
74        }
75    }
76}
77
78#[derive(Debug, Clone, Default)]
79struct OrderCorrelation {
80    customer_order_refs: AHashSet<CustomerOrderRef>,
81    strategy_id: Option<StrategyId>,
82    venue_order_id: Option<VenueOrderId>,
83    venue_order_ids: AHashSet<BetId>,
84    accepted: bool,
85    terminal_retained: bool,
86}
87
88#[derive(Debug, Clone)]
89enum TerminalRetentionKey {
90    Owned(ClientOrderId),
91    External(BetId),
92}
93
94/// Shared mutable state for the OCM stream handler.
95///
96/// Accessed by both the TCP reader closure and the execution client methods
97/// (submit, modify, connect/disconnect). All access goes through `Arc<Mutex<>>`.
98///
99/// Terminal retention owns order correlation, per-Bet fill and reduction state, and replacement
100/// history. The first REST, OCM, or reconciliation observation with active quantity at least the
101/// requested quantity and below the original confirms a pending reduction; later observations are
102/// no-ops.
103#[derive(Clone, Debug, Default)]
104pub struct OcmState {
105    /// Tracks cumulative per-bet fill and void state for deduplication and reconciliation.
106    pub fill_tracker: FillTracker,
107    /// Bet IDs that have received a terminal event (cancel, lapse, fill-complete).
108    pub terminal_orders: AHashSet<BetId>,
109    /// Old bet IDs from replace operations, to suppress late stream updates.
110    pub replaced_venue_order_ids: AHashSet<BetId>,
111    pub(crate) customer_order_refs: AHashMap<CustomerOrderRef, CustomerOrderRefResolution>,
112    order_correlations: AHashMap<ClientOrderId, OrderCorrelation>,
113    terminal_order_queue: VecDeque<TerminalRetentionKey>,
114    canceled_replace_bet_ids: AHashSet<BetId>,
115    pending_replace_state: AHashMap<(ClientOrderId, BetId), PendingReplaceState>,
116    pending_reductions: AHashMap<BetId, PendingReductionState>,
117}
118
119impl OcmState {
120    /// Bounds dedup memory while retaining recent delayed stream and REST overlap.
121    pub(crate) const DEDUP_RETENTION: usize = 10_000;
122
123    pub(crate) fn register_submission(
124        &mut self,
125        client_order_id: ClientOrderId,
126        strategy_id: StrategyId,
127    ) -> Result<(), String> {
128        self.register_order_ref(client_order_id)?;
129        self.register_order_identity(client_order_id, strategy_id);
130        Ok(())
131    }
132
133    /// Registers a customer_order_ref mapping for a new order.
134    pub fn register_customer_order_ref(&mut self, client_order_id: ClientOrderId) {
135        let _ = self.register_order_ref(client_order_id);
136    }
137
138    /// Registers both current and legacy customer_order_ref truncations.
139    pub fn register_customer_order_ref_with_legacy(&mut self, client_order_id: ClientOrderId) {
140        let current = make_customer_order_ref(client_order_id.as_str());
141        let legacy = make_customer_order_ref_legacy(client_order_id.as_str());
142        self.upsert_order_correlation(client_order_id, None, false, [current, legacy]);
143    }
144
145    /// Records the submitting strategy for a tracked order.
146    pub fn register_order_identity(
147        &mut self,
148        client_order_id: ClientOrderId,
149        strategy_id: StrategyId,
150    ) {
151        self.order_correlations
152            .entry(client_order_id)
153            .or_default()
154            .strategy_id = Some(strategy_id);
155    }
156
157    pub(crate) fn register_order_ref(
158        &mut self,
159        client_order_id: ClientOrderId,
160    ) -> Result<(), String> {
161        let customer_order_ref = make_customer_order_ref(client_order_id.as_str());
162
163        if self
164            .customer_order_refs
165            .get(&customer_order_ref)
166            .is_some_and(|resolution| {
167                *resolution != CustomerOrderRefResolution::Unique(client_order_id)
168            })
169        {
170            return Err(customer_order_ref);
171        }
172
173        self.upsert_order_correlation(client_order_id, None, false, [customer_order_ref]);
174        Ok(())
175    }
176
177    pub(crate) fn restore_order(
178        &mut self,
179        client_order_id: ClientOrderId,
180        strategy_id: StrategyId,
181        venue_order_id: VenueOrderId,
182    ) {
183        let current = make_customer_order_ref(client_order_id.as_str());
184        let legacy = make_customer_order_ref_legacy(client_order_id.as_str());
185        self.upsert_order_correlation(client_order_id, Some(strategy_id), true, [current, legacy]);
186        self.bind_venue_order_id(&client_order_id, venue_order_id);
187    }
188
189    fn upsert_order_correlation(
190        &mut self,
191        client_order_id: ClientOrderId,
192        strategy_id: Option<StrategyId>,
193        accepted: bool,
194        customer_order_refs: impl IntoIterator<Item = String>,
195    ) {
196        let correlation = self.order_correlations.entry(client_order_id).or_default();
197        if let Some(strategy_id) = strategy_id {
198            correlation.strategy_id = Some(strategy_id);
199        }
200
201        correlation.accepted |= accepted;
202        let mut affected = Vec::new();
203
204        for customer_order_ref in customer_order_refs {
205            if correlation
206                .customer_order_refs
207                .insert(customer_order_ref.clone())
208            {
209                affected.push(customer_order_ref);
210            }
211        }
212
213        for customer_order_ref in affected {
214            self.add_customer_order_ref(client_order_id, customer_order_ref);
215        }
216    }
217
218    fn add_customer_order_ref(
219        &mut self,
220        client_order_id: ClientOrderId,
221        customer_order_ref: String,
222    ) {
223        match self.customer_order_refs.get(&customer_order_ref).copied() {
224            None => {
225                self.customer_order_refs.insert(
226                    customer_order_ref,
227                    CustomerOrderRefResolution::Unique(client_order_id),
228                );
229            }
230            Some(CustomerOrderRefResolution::Unique(owner)) if owner != client_order_id => {
231                self.customer_order_refs
232                    .insert(customer_order_ref, CustomerOrderRefResolution::Ambiguous);
233            }
234            Some(_) => {}
235        }
236    }
237
238    /// Returns the submitting strategy for a tracked order, if known.
239    pub fn order_strategy_id(&self, client_order_id: &ClientOrderId) -> Option<StrategyId> {
240        self.order_correlations
241            .get(client_order_id)
242            .and_then(|correlation| correlation.strategy_id)
243    }
244
245    pub(crate) fn bind_venue_order_id(
246        &mut self,
247        client_order_id: &ClientOrderId,
248        venue_order_id: VenueOrderId,
249    ) {
250        let bet_id = venue_order_id.to_string();
251        let correlation = self.order_correlations.entry(*client_order_id).or_default();
252        let newly_correlated = correlation.venue_order_ids.insert(bet_id.clone());
253        correlation.venue_order_id = Some(venue_order_id);
254
255        if self.terminal_orders.contains(&bet_id) {
256            if newly_correlated {
257                self.remove_external_terminal_retention(&bet_id);
258            }
259
260            if !self.has_pending_replace(client_order_id) {
261                self.retain_terminal_identity(*client_order_id);
262            }
263        }
264    }
265
266    /// Records that acceptance has been emitted for a tracked order.
267    ///
268    /// Returns `true` when this call newly marks the order accepted (the caller
269    /// should emit `OrderAccepted`), or `false` when acceptance was already emitted.
270    pub fn mark_accepted(&mut self, client_order_id: ClientOrderId) -> bool {
271        let correlation = self.order_correlations.entry(client_order_id).or_default();
272        if correlation.accepted {
273            return false;
274        }
275
276        correlation.accepted = true;
277        true
278    }
279
280    pub(crate) fn is_accepted(&self, client_order_id: &ClientOrderId) -> bool {
281        self.order_correlations
282            .get(client_order_id)
283            .is_some_and(|correlation| correlation.accepted)
284    }
285
286    pub(crate) fn claim_acceptance(
287        &mut self,
288        client_order_id: ClientOrderId,
289        venue_order_id: VenueOrderId,
290    ) -> bool {
291        if !self.mark_accepted(client_order_id) {
292            return false;
293        }
294
295        self.bind_venue_order_id(&client_order_id, venue_order_id);
296        true
297    }
298
299    pub(crate) fn remove_order_correlation(&mut self, client_order_id: &ClientOrderId) {
300        self.remove_terminal_retention(client_order_id);
301        let Some(correlation) = self.order_correlations.remove(client_order_id) else {
302            return;
303        };
304        let OrderCorrelation {
305            customer_order_refs: affected,
306            venue_order_ids,
307            ..
308        } = correlation;
309
310        for customer_order_ref in affected {
311            if self
312                .customer_order_refs
313                .get(&customer_order_ref)
314                .is_some_and(|resolution| {
315                    *resolution == CustomerOrderRefResolution::Unique(*client_order_id)
316                })
317            {
318                self.customer_order_refs.remove(&customer_order_ref);
319            } else {
320                self.rebuild_customer_order_ref(&customer_order_ref);
321            }
322        }
323
324        self.pending_reductions
325            .retain(|_, pending| pending.client_order_id != *client_order_id);
326        self.pending_replace_state
327            .retain(|(candidate, _), _| candidate != client_order_id);
328
329        for bet_id in venue_order_ids {
330            self.remove_venue_order_state(&bet_id);
331        }
332    }
333
334    /// Removes customer_order_ref mappings for a client_order_id.
335    pub fn remove_customer_order_refs(&mut self, client_order_id: &ClientOrderId) {
336        self.remove_order_correlation(client_order_id);
337    }
338
339    fn rebuild_customer_order_ref(&mut self, customer_order_ref: &str) {
340        let mut owners = self
341            .order_correlations
342            .iter()
343            .filter(|(_, correlation)| correlation.customer_order_refs.contains(customer_order_ref))
344            .map(|(client_order_id, _)| *client_order_id);
345        let resolution = owners.next().map(|client_order_id| {
346            if owners.next().is_some() {
347                CustomerOrderRefResolution::Ambiguous
348            } else {
349                CustomerOrderRefResolution::Unique(client_order_id)
350            }
351        });
352
353        if let Some(resolution) = resolution {
354            self.customer_order_refs
355                .insert(customer_order_ref.to_string(), resolution);
356        } else {
357            self.customer_order_refs.remove(customer_order_ref);
358        }
359    }
360
361    pub(crate) fn customer_order_ref_resolution(
362        &self,
363        customer_order_ref: &str,
364    ) -> Option<CustomerOrderRefResolution> {
365        self.customer_order_refs.get(customer_order_ref).copied()
366    }
367
368    pub(crate) fn resolve_order_owner(
369        &self,
370        customer_order_ref: Option<&str>,
371        venue_order_id: &str,
372    ) -> Option<CustomerOrderRefResolution> {
373        customer_order_ref
374            .and_then(|reference| self.customer_order_ref_resolution(reference))
375            .or_else(|| {
376                self.client_order_id_by_venue_order_id(venue_order_id)
377                    .map(CustomerOrderRefResolution::Unique)
378            })
379    }
380
381    pub(crate) fn client_order_id_by_venue_order_id(
382        &self,
383        venue_order_id: &str,
384    ) -> Option<ClientOrderId> {
385        let mut owners = self
386            .order_correlations
387            .iter()
388            .filter(|(_, correlation)| correlation.venue_order_ids.contains(venue_order_id))
389            .map(|(client_order_id, _)| *client_order_id);
390        let owner = owners.next()?;
391        owners.next().is_none().then_some(owner)
392    }
393
394    /// Resolves a client_order_id from the unmatched order's rfo field.
395    pub fn resolve_client_order_id(&self, rfo: Option<&str>) -> Option<ClientOrderId> {
396        rfo.and_then(|customer_order_ref| {
397            self.customer_order_ref_resolution(customer_order_ref)
398                .and_then(CustomerOrderRefResolution::client_order_id)
399        })
400    }
401
402    /// Returns `true` if a cancel/lapse for this bet should be suppressed
403    /// because a replace operation is pending or the bet was already replaced.
404    pub fn should_suppress_cancel(&self, client_order_id: &ClientOrderId, bet_id: &str) -> bool {
405        if self.replaced_venue_order_ids.contains(bet_id) {
406            return true;
407        }
408
409        self.pending_replace_state
410            .contains_key(&(*client_order_id, bet_id.to_string()))
411    }
412
413    pub(crate) fn is_retained_terminal_order(&self, client_order_id: &ClientOrderId) -> bool {
414        self.order_correlations
415            .get(client_order_id)
416            .is_some_and(|correlation| correlation.terminal_retained)
417    }
418
419    pub(crate) fn should_suppress_replaced_report(&self, bet_id: &str) -> bool {
420        if !self.replaced_venue_order_ids.contains(bet_id) {
421            return false;
422        }
423
424        self.client_order_id_by_venue_order_id(bet_id)
425            .is_none_or(|client_order_id| !self.is_retained_terminal_order(&client_order_id))
426    }
427
428    pub(crate) fn register_pending_replace(
429        &mut self,
430        client_order_id: ClientOrderId,
431        old_bet_id: String,
432        total_quantity: Option<Quantity>,
433    ) {
434        self.remove_terminal_retention(&client_order_id);
435        let key = (client_order_id, old_bet_id);
436
437        self.pending_replace_state.insert(
438            key,
439            PendingReplaceState {
440                id: UUID4::new(),
441                total_quantity,
442                awaiting_reconciliation: false,
443            },
444        );
445    }
446
447    pub(crate) fn mark_pending_replace_ambiguous(
448        &mut self,
449        client_order_id: ClientOrderId,
450        old_bet_id: &str,
451    ) {
452        let key = (client_order_id, old_bet_id.to_string());
453
454        if let Some(pending) = self.pending_replace_state.get_mut(&key) {
455            pending.awaiting_reconciliation = true;
456        }
457    }
458
459    pub(crate) fn pending_replace_awaits_reconciliation(
460        &self,
461        client_order_id: &ClientOrderId,
462        old_bet_id: &str,
463    ) -> bool {
464        self.pending_replace_state
465            .get(&(*client_order_id, old_bet_id.to_string()))
466            .is_some_and(|pending| pending.awaiting_reconciliation)
467    }
468
469    pub(crate) fn pending_replace_reconciliation_ids(
470        &self,
471    ) -> AHashMap<(ClientOrderId, BetId), UUID4> {
472        self.pending_replace_state
473            .iter()
474            .filter(|(_, pending)| pending.awaiting_reconciliation)
475            .map(|(key, pending)| (key.clone(), pending.id))
476            .collect()
477    }
478
479    pub(crate) fn take_pending_replace(
480        &mut self,
481        client_order_id: ClientOrderId,
482        old_bet_id: &str,
483    ) -> Option<PendingReplaceState> {
484        let key = (client_order_id, old_bet_id.to_string());
485        self.pending_replace_state.remove(&key)
486    }
487
488    fn has_pending_replace(&self, client_order_id: &ClientOrderId) -> bool {
489        self.pending_replace_state
490            .keys()
491            .any(|(candidate, _)| candidate == client_order_id)
492    }
493
494    pub(crate) fn order_status_supplemental_ids(
495        &self,
496        mut client_order_ids: AHashSet<ClientOrderId>,
497    ) -> (AHashSet<BetId>, AHashMap<CustomerOrderRef, AHashSet<BetId>>) {
498        let mut customer_order_refs = AHashMap::new();
499
500        for (client_order_id, bet_id) in self.pending_replace_state.keys() {
501            client_order_ids.insert(*client_order_id);
502            if let Some(correlation) = self.order_correlations.get(client_order_id) {
503                for customer_order_ref in &correlation.customer_order_refs {
504                    customer_order_refs
505                        .entry(customer_order_ref.clone())
506                        .or_insert_with(AHashSet::new)
507                        .insert(bet_id.clone());
508                }
509            }
510        }
511
512        client_order_ids.extend(
513            self.pending_reductions
514                .values()
515                .filter(|pending| pending.confirmed_quantity.is_none())
516                .map(|pending| pending.client_order_id),
517        );
518        let bet_ids = client_order_ids
519            .iter()
520            .filter_map(|client_order_id| self.order_correlations.get(client_order_id))
521            .flat_map(|correlation| correlation.venue_order_ids.iter().cloned())
522            .chain(
523                self.pending_replace_state
524                    .keys()
525                    .map(|(_, bet_id)| bet_id.clone()),
526            )
527            .chain(
528                self.pending_reductions
529                    .iter()
530                    .filter(|(_, pending)| pending.confirmed_quantity.is_none())
531                    .map(|(bet_id, _)| bet_id.clone()),
532            )
533            .collect();
534        (bet_ids, customer_order_refs)
535    }
536
537    /// Returns `true` when `bet_id` is the order's most recently placed Bet.
538    pub(crate) fn is_current_venue_order_id(
539        &self,
540        client_order_id: &ClientOrderId,
541        bet_id: &str,
542    ) -> bool {
543        self.order_correlations
544            .get(client_order_id)
545            .and_then(|correlation| correlation.venue_order_id.as_ref())
546            .is_some_and(|venue_order_id| venue_order_id.as_str() == bet_id)
547    }
548
549    pub(crate) fn mark_canceled_replace(&mut self, client_order_id: ClientOrderId, bet_id: &str) {
550        self.canceled_replace_bet_ids.insert(bet_id.to_string());
551        self.retain_terminal_order(client_order_id, bet_id);
552    }
553
554    pub(crate) fn is_canceled_replace(&self, bet_id: &str) -> bool {
555        self.canceled_replace_bet_ids.contains(bet_id)
556    }
557
558    pub(crate) fn is_redundant_terminal_update(&self, order: &UnmatchedOrder) -> bool {
559        self.terminal_orders.contains(&order.id)
560            && !self.is_canceled_replace(&order.id)
561            && !self.fill_tracker.has_unseen_fill(order)
562            && !self.fill_tracker.has_unseen_fill_void(order)
563            && !self.fill_tracker.has_unseen_price_restatement(order)
564    }
565
566    pub(crate) fn clear_canceled_replace(&mut self, bet_id: &str) {
567        self.canceled_replace_bet_ids.remove(bet_id);
568    }
569
570    /// Promotes a new bet observed while a price replacement is pending.
571    ///
572    /// Returns the total order quantity and old Bet ID when `new_bet_id` differs from one pending
573    /// old Bet ID.
574    pub(crate) fn promote_pending_replace(
575        &mut self,
576        client_order_id: &ClientOrderId,
577        new_bet_id: &str,
578        replacement_quantity: Quantity,
579    ) -> Option<(Quantity, String)> {
580        let mut old_bet_ids = self
581            .pending_replace_state
582            .keys()
583            .filter(|(candidate, old_bet_id)| {
584                candidate == client_order_id && old_bet_id != new_bet_id
585            })
586            .map(|(_, old_bet_id)| old_bet_id.clone());
587
588        let old_bet_id = old_bet_ids.next()?;
589        if old_bet_ids.next().is_some() {
590            return None;
591        }
592
593        let pending = self.complete_pending_replace(
594            *client_order_id,
595            &old_bet_id,
596            VenueOrderId::from(new_bet_id),
597        )?;
598
599        Some((
600            pending.total_quantity.unwrap_or(replacement_quantity),
601            old_bet_id,
602        ))
603    }
604
605    pub(crate) fn complete_pending_replace(
606        &mut self,
607        client_order_id: ClientOrderId,
608        old_bet_id: &str,
609        new_venue_order_id: VenueOrderId,
610    ) -> Option<PendingReplaceState> {
611        if self
612            .replaced_venue_order_ids
613            .contains(new_venue_order_id.as_str())
614        {
615            return None;
616        }
617
618        let pending = self.take_pending_replace(client_order_id, old_bet_id)?;
619        self.mark_replaced_venue_order_id(client_order_id, old_bet_id.to_string());
620        self.bind_venue_order_id(&client_order_id, new_venue_order_id);
621        Some(pending)
622    }
623
624    fn mark_replaced_venue_order_id(&mut self, client_order_id: ClientOrderId, bet_id: String) {
625        self.mark_correlated_terminal_order(client_order_id, &bet_id);
626        self.replaced_venue_order_ids.insert(bet_id);
627    }
628
629    pub(crate) fn register_pending_reduction(
630        &mut self,
631        client_order_id: ClientOrderId,
632        bet_id: String,
633        original_quantity: Quantity,
634        requested_quantity: Quantity,
635    ) {
636        self.pending_reductions.insert(
637            bet_id,
638            PendingReductionState {
639                client_order_id,
640                original_quantity,
641                requested_quantity,
642                confirmed_quantity: None,
643            },
644        );
645    }
646
647    pub(crate) fn replaced_matched_quantity(&self, client_order_id: &ClientOrderId) -> Decimal {
648        self.order_correlations
649            .get(client_order_id)
650            .into_iter()
651            .flat_map(|correlation| &correlation.venue_order_ids)
652            .filter(|bet_id| self.replaced_venue_order_ids.contains(*bet_id))
653            .map(|bet_id| self.fill_tracker.matched_quantity(bet_id))
654            .sum()
655    }
656
657    pub(crate) fn replaced_voided_quantity(&self, client_order_id: &ClientOrderId) -> Decimal {
658        self.order_correlations
659            .get(client_order_id)
660            .into_iter()
661            .flat_map(|correlation| &correlation.venue_order_ids)
662            .filter(|bet_id| self.replaced_venue_order_ids.contains(*bet_id))
663            .map(|bet_id| self.fill_tracker.voided_quantity(bet_id))
664            .sum()
665    }
666
667    pub(crate) fn confirm_pending_reduction(
668        &mut self,
669        client_order_id: &ClientOrderId,
670        bet_id: &str,
671        active_quantity: Quantity,
672    ) -> Option<Quantity> {
673        let pending = self.pending_reductions.get_mut(bet_id)?;
674
675        if !pending.can_confirm(client_order_id, active_quantity) {
676            return None;
677        }
678
679        pending.confirmed_quantity = Some(active_quantity);
680        Some(active_quantity)
681    }
682
683    pub(crate) fn complete_pending_reduction(
684        &mut self,
685        client_order_id: &ClientOrderId,
686        bet_id: &str,
687        quantity: Quantity,
688    ) -> bool {
689        let Some(pending) = self.pending_reductions.get_mut(bet_id) else {
690            return false;
691        };
692
693        if !pending.is_unconfirmed_for(client_order_id) {
694            return false;
695        }
696
697        pending.confirmed_quantity = Some(quantity);
698        true
699    }
700
701    pub(crate) fn reduced_quantity(&self, bet_id: &str) -> Option<Quantity> {
702        self.pending_reductions
703            .get(bet_id)
704            .and_then(|pending| pending.confirmed_quantity)
705    }
706
707    pub(crate) fn clear_pending_reduction(
708        &mut self,
709        client_order_id: &ClientOrderId,
710        bet_id: &str,
711    ) {
712        if self
713            .pending_reductions
714            .get(bet_id)
715            .is_some_and(|pending| pending.client_order_id == *client_order_id)
716        {
717            self.pending_reductions.remove(bet_id);
718        }
719    }
720
721    /// Retains a locally owned terminal identity and all of its Bet ID state.
722    pub(crate) fn retain_terminal_order(&mut self, client_order_id: ClientOrderId, bet_id: &str) {
723        if self.replaced_venue_order_ids.contains(bet_id)
724            || self.has_pending_replace(&client_order_id)
725        {
726            self.mark_correlated_terminal_order(client_order_id, bet_id);
727            return;
728        }
729
730        self.bind_venue_order_id(&client_order_id, VenueOrderId::from(bet_id));
731        self.mark_correlated_terminal_order(client_order_id, bet_id);
732
733        self.retain_terminal_identity(client_order_id);
734    }
735
736    fn mark_correlated_terminal_order(&mut self, client_order_id: ClientOrderId, bet_id: &str) {
737        if self
738            .pending_reductions
739            .get(bet_id)
740            .is_some_and(|pending| pending.is_unconfirmed_for(&client_order_id))
741        {
742            self.pending_reductions.remove(bet_id);
743        }
744
745        let newly_correlated = self
746            .order_correlations
747            .entry(client_order_id)
748            .or_default()
749            .venue_order_ids
750            .insert(bet_id.to_string());
751        if newly_correlated {
752            self.remove_external_terminal_retention(bet_id);
753        }
754        self.terminal_orders.insert(bet_id.to_string());
755    }
756
757    fn retain_terminal_identity(&mut self, client_order_id: ClientOrderId) {
758        let correlation = self.order_correlations.entry(client_order_id).or_default();
759        if correlation.terminal_retained {
760            return;
761        }
762
763        correlation.terminal_retained = true;
764        self.push_terminal_identity(TerminalRetentionKey::Owned(client_order_id));
765    }
766
767    /// Records an external terminal bet and bounds its stream and REST dedup state.
768    pub fn mark_terminal_order(&mut self, bet_id: String) {
769        if !self.terminal_orders.insert(bet_id.clone()) {
770            return;
771        }
772
773        self.push_terminal_identity(TerminalRetentionKey::External(bet_id));
774    }
775
776    fn push_terminal_identity(&mut self, key: TerminalRetentionKey) {
777        self.terminal_order_queue.push_back(key);
778        if self.terminal_order_queue.len() > Self::DEDUP_RETENTION
779            && let Some(expired) = self.terminal_order_queue.pop_front()
780        {
781            self.evict_terminal_identity(expired);
782        }
783    }
784
785    fn remove_terminal_retention(&mut self, client_order_id: &ClientOrderId) {
786        if self
787            .order_correlations
788            .get_mut(client_order_id)
789            .is_some_and(|correlation| std::mem::take(&mut correlation.terminal_retained))
790        {
791            self.terminal_order_queue.retain(
792                |key| !matches!(key, TerminalRetentionKey::Owned(candidate) if candidate == client_order_id),
793            );
794        }
795    }
796
797    fn remove_external_terminal_retention(&mut self, bet_id: &str) {
798        if !self.terminal_orders.contains(bet_id) {
799            return;
800        }
801
802        self.terminal_order_queue.retain(
803            |key| !matches!(key, TerminalRetentionKey::External(candidate) if candidate == bet_id),
804        );
805    }
806
807    pub(crate) fn mark_order_active(&mut self, client_order_id: &ClientOrderId, bet_id: &str) {
808        self.remove_terminal_retention(client_order_id);
809        self.terminal_orders.remove(bet_id);
810        self.canceled_replace_bet_ids.remove(bet_id);
811    }
812
813    fn evict_terminal_identity(&mut self, key: TerminalRetentionKey) {
814        match key {
815            TerminalRetentionKey::Owned(client_order_id) => {
816                let retained = self
817                    .order_correlations
818                    .get_mut(&client_order_id)
819                    .is_some_and(|correlation| std::mem::take(&mut correlation.terminal_retained));
820                if retained {
821                    self.remove_order_correlation(&client_order_id);
822                }
823            }
824            TerminalRetentionKey::External(bet_id) => {
825                self.remove_venue_order_state(&bet_id);
826            }
827        }
828    }
829
830    fn remove_venue_order_state(&mut self, bet_id: &str) {
831        self.terminal_orders.remove(bet_id);
832        self.replaced_venue_order_ids.remove(bet_id);
833        self.canceled_replace_bet_ids.remove(bet_id);
834        self.pending_reductions.remove(bet_id);
835        self.fill_tracker.prune(bet_id);
836    }
837
838    /// Anchors the fill tracker against cached orders so the post-reconnect
839    /// image neither treats cumulative size as a new fill nor re-emits a
840    /// fill that was published via another channel.
841    pub fn sync_from_orders(&mut self, orders: &[OrderSyncEntry]) {
842        for entry in orders {
843            self.restore_order(
844                entry.client_order_id,
845                entry.strategy_id,
846                VenueOrderId::from(entry.bet_id.as_str()),
847            );
848
849            for venue_order_id in &entry.venue_order_ids {
850                if venue_order_id != &entry.bet_id {
851                    self.mark_replaced_venue_order_id(
852                        entry.client_order_id,
853                        venue_order_id.clone(),
854                    );
855                }
856            }
857
858            if entry.is_closed {
859                self.retain_terminal_order(entry.client_order_id, &entry.bet_id);
860            } else {
861                self.mark_order_active(&entry.client_order_id, &entry.bet_id);
862            }
863
864            if entry.filled_qty > Decimal::ZERO {
865                self.fill_tracker
866                    .sync_order(&entry.bet_id, entry.filled_qty, entry.avg_px);
867            }
868
869            if !entry.trade_ids.is_empty() {
870                self.fill_tracker
871                    .seed_published_trade_ids(entry.trade_ids.iter().cloned());
872            }
873        }
874    }
875}
876
877#[cfg(test)]
878mod tests {
879    use rstest::rstest;
880
881    use super::*;
882
883    #[rstest]
884    fn owned_terminal_retention_evicts_all_identity_state_together() {
885        let mut state = OcmState::default();
886        let client_order_id = ClientOrderId::from("O-TERMINAL-0");
887        let strategy_id = StrategyId::from("S-001");
888        let current_bet_id = "current-bet-0";
889        let replaced_bet_id = "replaced-bet-0";
890        let size_matched = Decimal::from(2);
891        let average_price = Decimal::from(3);
892        state.restore_order(
893            client_order_id,
894            strategy_id,
895            VenueOrderId::from(current_bet_id),
896        );
897        state.mark_replaced_venue_order_id(client_order_id, replaced_bet_id.to_string());
898        assert!(state.is_accepted(&client_order_id));
899        state.register_pending_reduction(
900            client_order_id,
901            current_bet_id.to_string(),
902            Quantity::from(10),
903            Quantity::from(4),
904        );
905        state.confirm_pending_reduction(&client_order_id, current_bet_id, Quantity::from(4));
906        state.fill_tracker.advance_cumulative_fill(
907            current_bet_id,
908            size_matched,
909            Some(average_price),
910            average_price,
911        );
912        state.fill_tracker.advance_cumulative_fill(
913            replaced_bet_id,
914            size_matched,
915            Some(average_price),
916            average_price,
917        );
918        state.retain_terminal_order(client_order_id, current_bet_id);
919
920        for index in 0..OcmState::DEDUP_RETENTION {
921            state.mark_terminal_order(format!("external-bet-{index}"));
922        }
923
924        let current_replay = state.fill_tracker.advance_cumulative_fill(
925            current_bet_id,
926            size_matched,
927            Some(average_price),
928            average_price,
929        );
930        let replaced_replay = state.fill_tracker.advance_cumulative_fill(
931            replaced_bet_id,
932            size_matched,
933            Some(average_price),
934            average_price,
935        );
936        let customer_order_ref = make_customer_order_ref(client_order_id.as_str());
937
938        assert_eq!(state.order_strategy_id(&client_order_id), None);
939        assert_eq!(
940            state.resolve_client_order_id(Some(&customer_order_ref)),
941            None,
942        );
943        assert_eq!(state.reduced_quantity(current_bet_id), None);
944        assert!(!state.terminal_orders.contains(current_bet_id));
945        assert!(!state.terminal_orders.contains(replaced_bet_id));
946        assert!(!state.replaced_venue_order_ids.contains(replaced_bet_id));
947        assert!(current_replay.is_some());
948        assert!(replaced_replay.is_some());
949    }
950
951    #[rstest]
952    fn terminal_retention_does_not_evict_active_or_ambiguous_replace_identity() {
953        let mut state = OcmState::default();
954        let active_client_order_id = ClientOrderId::from("O-ACTIVE");
955        let ambiguous_client_order_id = ClientOrderId::from("O-AMBIGUOUS");
956        let strategy_id = StrategyId::from("S-001");
957        state.restore_order(
958            active_client_order_id,
959            strategy_id,
960            VenueOrderId::from("active-bet"),
961        );
962        state.restore_order(
963            ambiguous_client_order_id,
964            strategy_id,
965            VenueOrderId::from("ambiguous-bet"),
966        );
967        state.register_pending_replace(
968            ambiguous_client_order_id,
969            "ambiguous-bet".to_string(),
970            Some(Quantity::from(10)),
971        );
972        state.mark_pending_replace_ambiguous(ambiguous_client_order_id, "ambiguous-bet");
973        state.retain_terminal_order(ambiguous_client_order_id, "ambiguous-bet");
974
975        for index in 0..=OcmState::DEDUP_RETENTION {
976            state.mark_terminal_order(format!("external-bet-{index}"));
977        }
978
979        assert_eq!(
980            state.order_strategy_id(&active_client_order_id),
981            Some(strategy_id),
982        );
983        assert!(!state.mark_accepted(active_client_order_id));
984        assert_eq!(
985            state.order_strategy_id(&ambiguous_client_order_id),
986            Some(strategy_id),
987        );
988        assert!(state.pending_replace_awaits_reconciliation(
989            &ambiguous_client_order_id,
990            "ambiguous-bet",
991        ));
992        assert!(state.should_suppress_cancel(&ambiguous_client_order_id, "ambiguous-bet",));
993    }
994
995    #[rstest]
996    fn test_order_status_supplemental_ids_include_current_and_legacy_pending_replace_refs() {
997        let mut state = OcmState::default();
998        let client_order_id =
999            ClientOrderId::from("O-20240101-550e8400-e29b-41d4-a716-446655440000");
1000        let strategy_id = StrategyId::from("S-001");
1001        let old_bet_id = "old-bet";
1002        let current_ref = make_customer_order_ref(client_order_id.as_str());
1003        let legacy_ref = make_customer_order_ref_legacy(client_order_id.as_str());
1004        state.restore_order(client_order_id, strategy_id, VenueOrderId::from(old_bet_id));
1005        state.register_pending_replace(
1006            client_order_id,
1007            old_bet_id.to_string(),
1008            Some(Quantity::from(10)),
1009        );
1010
1011        let (bet_ids, customer_order_refs) = state.order_status_supplemental_ids(AHashSet::new());
1012
1013        assert_ne!(current_ref, legacy_ref);
1014        assert_eq!(bet_ids, AHashSet::from_iter([old_bet_id.to_string()]));
1015        assert_eq!(
1016            customer_order_refs,
1017            AHashMap::from_iter([
1018                (current_ref, AHashSet::from_iter([old_bet_id.to_string()])),
1019                (legacy_ref, AHashSet::from_iter([old_bet_id.to_string()])),
1020            ]),
1021        );
1022    }
1023
1024    #[rstest]
1025    fn successful_replace_without_old_leg_ocm_uses_terminal_identity_lifecycle() {
1026        let mut state = OcmState::default();
1027        let client_order_id = ClientOrderId::from("O-REPLACE");
1028        let strategy_id = StrategyId::from("S-001");
1029        state.restore_order(client_order_id, strategy_id, VenueOrderId::from("old-bet"));
1030        state.register_pending_replace(
1031            client_order_id,
1032            "old-bet".to_string(),
1033            Some(Quantity::from(10)),
1034        );
1035
1036        let completed = state.complete_pending_replace(
1037            client_order_id,
1038            "old-bet",
1039            VenueOrderId::from("new-bet"),
1040        );
1041
1042        assert!(completed.is_some());
1043        assert!(state.pending_replace_state.is_empty());
1044        assert!(state.replaced_venue_order_ids.contains("old-bet"));
1045        assert!(state.terminal_orders.contains("old-bet"));
1046        assert_eq!(
1047            state.client_order_id_by_venue_order_id("old-bet"),
1048            Some(client_order_id),
1049        );
1050        assert_eq!(
1051            state.client_order_id_by_venue_order_id("new-bet"),
1052            Some(client_order_id),
1053        );
1054        assert!(!state.is_retained_terminal_order(&client_order_id));
1055
1056        state.retain_terminal_order(client_order_id, "old-bet");
1057
1058        assert_eq!(
1059            state
1060                .order_correlations
1061                .get(&client_order_id)
1062                .and_then(|correlation| correlation.venue_order_id),
1063            Some(VenueOrderId::from("new-bet")),
1064        );
1065        assert!(!state.is_retained_terminal_order(&client_order_id));
1066
1067        state.retain_terminal_order(client_order_id, "new-bet");
1068        state.retain_terminal_order(client_order_id, "old-bet");
1069
1070        assert_eq!(
1071            state
1072                .order_correlations
1073                .get(&client_order_id)
1074                .and_then(|correlation| correlation.venue_order_id),
1075            Some(VenueOrderId::from("new-bet")),
1076        );
1077        assert_eq!(state.terminal_order_queue.len(), 1);
1078    }
1079
1080    #[rstest]
1081    fn replace_resolution_does_not_discard_another_pending_mutation() {
1082        let mut state = OcmState::default();
1083        let client_order_id = ClientOrderId::from("O-REPLACE");
1084        state.restore_order(
1085            client_order_id,
1086            StrategyId::from("S-001"),
1087            VenueOrderId::from("old-bet-1"),
1088        );
1089        state.register_pending_replace(
1090            client_order_id,
1091            "old-bet-1".to_string(),
1092            Some(Quantity::from(10)),
1093        );
1094        state.register_pending_replace(
1095            client_order_id,
1096            "old-bet-2".to_string(),
1097            Some(Quantity::from(10)),
1098        );
1099
1100        let unresolved_promotion =
1101            state.promote_pending_replace(&client_order_id, "new-bet", Quantity::from(10));
1102        let completed = state.complete_pending_replace(
1103            client_order_id,
1104            "old-bet-1",
1105            VenueOrderId::from("new-bet"),
1106        );
1107
1108        assert_eq!(unresolved_promotion, None);
1109        assert!(completed.is_some());
1110        assert!(
1111            !state
1112                .pending_replace_state
1113                .contains_key(&(client_order_id, "old-bet-1".to_string()))
1114        );
1115        assert!(
1116            state
1117                .pending_replace_state
1118                .contains_key(&(client_order_id, "old-bet-2".to_string()))
1119        );
1120        assert!(state.replaced_venue_order_ids.contains("old-bet-1"));
1121        assert!(!state.replaced_venue_order_ids.contains("old-bet-2"));
1122        assert_eq!(
1123            state.client_order_id_by_venue_order_id("new-bet"),
1124            Some(client_order_id),
1125        );
1126    }
1127
1128    #[rstest]
1129    fn historical_bet_migrates_from_external_to_owned_identity() {
1130        let mut state = OcmState::default();
1131        let client_order_id = ClientOrderId::from("O-MIGRATED");
1132        let old_bet_id = "old-bet";
1133        let size_matched = Decimal::from(2);
1134        let average_price = Decimal::from(3);
1135        state.mark_terminal_order(old_bet_id.to_string());
1136        state.fill_tracker.advance_cumulative_fill(
1137            old_bet_id,
1138            size_matched,
1139            Some(average_price),
1140            average_price,
1141        );
1142        state.restore_order(
1143            client_order_id,
1144            StrategyId::from("S-001"),
1145            VenueOrderId::from("current-bet"),
1146        );
1147        state.mark_replaced_venue_order_id(client_order_id, old_bet_id.to_string());
1148
1149        for index in 0..=OcmState::DEDUP_RETENTION {
1150            state.mark_terminal_order(format!("external-bet-{index}"));
1151        }
1152
1153        let replay = state.fill_tracker.advance_cumulative_fill(
1154            old_bet_id,
1155            size_matched,
1156            Some(average_price),
1157            average_price,
1158        );
1159
1160        assert_eq!(
1161            state.client_order_id_by_venue_order_id(old_bet_id),
1162            Some(client_order_id),
1163        );
1164        assert!(state.terminal_orders.contains(old_bet_id));
1165        assert!(state.replaced_venue_order_ids.contains(old_bet_id));
1166        assert!(replay.is_none());
1167    }
1168
1169    #[rstest]
1170    fn sustained_owned_terminal_history_stays_bounded() {
1171        let mut state = OcmState::default();
1172        let strategy_id = StrategyId::from("S-001");
1173
1174        for index in 0..=OcmState::DEDUP_RETENTION {
1175            let client_order_id = ClientOrderId::from(format!("O-{index}"));
1176            let current_bet_id = format!("current-bet-{index}");
1177            state.restore_order(
1178                client_order_id,
1179                strategy_id,
1180                VenueOrderId::from(current_bet_id.as_str()),
1181            );
1182            state.mark_replaced_venue_order_id(client_order_id, format!("replaced-bet-{index}"));
1183            state.retain_terminal_order(client_order_id, &current_bet_id);
1184        }
1185
1186        assert_eq!(state.order_correlations.len(), OcmState::DEDUP_RETENTION);
1187        assert_eq!(
1188            state
1189                .order_correlations
1190                .values()
1191                .filter(|correlation| correlation.terminal_retained)
1192                .count(),
1193            OcmState::DEDUP_RETENTION,
1194        );
1195        assert_eq!(state.terminal_order_queue.len(), OcmState::DEDUP_RETENTION);
1196        assert_eq!(
1197            state.replaced_venue_order_ids.len(),
1198            OcmState::DEDUP_RETENTION,
1199        );
1200        assert_eq!(state.terminal_orders.len(), OcmState::DEDUP_RETENTION * 2);
1201        assert_eq!(state.order_strategy_id(&ClientOrderId::from("O-0")), None,);
1202        assert_eq!(
1203            state.order_strategy_id(&ClientOrderId::from(format!(
1204                "O-{}",
1205                OcmState::DEDUP_RETENTION
1206            ))),
1207            Some(strategy_id),
1208        );
1209    }
1210
1211    #[rstest]
1212    fn terminal_order_retention_evicts_fill_tracker_state() {
1213        let mut state = OcmState::default();
1214        let first_bet_id = "bet-0";
1215        let size_matched = Decimal::new(10, 0);
1216        let average_price = Decimal::new(20, 1);
1217
1218        assert!(
1219            state
1220                .fill_tracker
1221                .advance_cumulative_fill(
1222                    first_bet_id,
1223                    size_matched,
1224                    Some(average_price),
1225                    average_price,
1226                )
1227                .is_some(),
1228        );
1229        state
1230            .replaced_venue_order_ids
1231            .insert(first_bet_id.to_string());
1232        state.mark_terminal_order(first_bet_id.to_string());
1233        for index in 1..=OcmState::DEDUP_RETENTION {
1234            state.mark_terminal_order(format!("bet-{index}"));
1235        }
1236
1237        let replay_after_eviction = state.fill_tracker.advance_cumulative_fill(
1238            first_bet_id,
1239            size_matched,
1240            Some(average_price),
1241            average_price,
1242        );
1243
1244        assert!(!state.terminal_orders.contains(first_bet_id));
1245        assert!(!state.replaced_venue_order_ids.contains(first_bet_id));
1246        assert!(replay_after_eviction.is_some());
1247    }
1248
1249    #[rstest]
1250    fn active_submission_collision_is_rejected() {
1251        let suffix = "12345678901234567890123456789012";
1252        let first = ClientOrderId::from(format!("FIRST-{suffix}"));
1253        let second = ClientOrderId::from(format!("SECOND-{suffix}"));
1254        let strategy_id = StrategyId::from("S-001");
1255        let mut state = OcmState::default();
1256
1257        state.register_submission(first, strategy_id).unwrap();
1258        let collision = state.register_submission(second, strategy_id);
1259
1260        assert_eq!(collision, Err(suffix.to_string()));
1261        assert_eq!(state.resolve_client_order_id(Some(suffix)), Some(first));
1262        assert_eq!(state.order_strategy_id(&second), None);
1263    }
1264
1265    #[rstest]
1266    fn submission_collision_with_restored_legacy_reference_is_rejected() {
1267        let reference = "12345678901234567890123456789012";
1268        let restored = ClientOrderId::from(format!("{reference}-RESTORED"));
1269        let fresh = ClientOrderId::from(format!("FRESH-{reference}"));
1270        let strategy_id = StrategyId::from("S-001");
1271        let mut state = OcmState::default();
1272
1273        state.restore_order(restored, strategy_id, VenueOrderId::from("bet-restored"));
1274        let collision = state.register_submission(fresh, strategy_id);
1275
1276        assert_eq!(collision, Err(reference.to_string()));
1277        assert_eq!(
1278            state.resolve_client_order_id(Some(reference)),
1279            Some(restored),
1280        );
1281        assert_eq!(state.order_strategy_id(&fresh), None);
1282    }
1283
1284    #[rstest]
1285    fn restored_current_reference_ambiguity_recovers_after_cleanup() {
1286        let suffix = "12345678901234567890123456789012";
1287        let first = ClientOrderId::from(format!("FIRST-{suffix}"));
1288        let second = ClientOrderId::from(format!("SECOND-{suffix}"));
1289        let strategy_id = StrategyId::from("S-001");
1290        let mut state = OcmState::default();
1291
1292        state.restore_order(first, strategy_id, VenueOrderId::from("bet-1"));
1293        state.restore_order(second, strategy_id, VenueOrderId::from("bet-2"));
1294
1295        assert_eq!(
1296            state.customer_order_ref_resolution(suffix),
1297            Some(CustomerOrderRefResolution::Ambiguous),
1298        );
1299        assert_eq!(state.resolve_client_order_id(Some(suffix)), None);
1300        state.remove_order_correlation(&first);
1301        assert_eq!(state.resolve_client_order_id(Some(suffix)), Some(second));
1302        assert_eq!(
1303            state
1304                .order_correlations
1305                .get(&second)
1306                .and_then(|correlation| correlation.venue_order_id),
1307            Some(VenueOrderId::from("bet-2")),
1308        );
1309        assert!(!state.mark_accepted(second));
1310    }
1311
1312    #[rstest]
1313    fn restored_legacy_reference_ambiguity_recovers_after_cleanup() {
1314        let prefix = "12345678901234567890123456789012";
1315        let first = ClientOrderId::from(format!("{prefix}-FIRST"));
1316        let second = ClientOrderId::from(format!("{prefix}-SECOND"));
1317        let strategy_id = StrategyId::from("S-001");
1318        let mut state = OcmState::default();
1319
1320        state.restore_order(first, strategy_id, VenueOrderId::from("bet-1"));
1321        state.restore_order(second, strategy_id, VenueOrderId::from("bet-2"));
1322
1323        assert_eq!(
1324            state.customer_order_ref_resolution(prefix),
1325            Some(CustomerOrderRefResolution::Ambiguous),
1326        );
1327        assert_eq!(state.resolve_client_order_id(Some(prefix)), None);
1328        state.remove_order_correlation(&second);
1329        assert_eq!(state.resolve_client_order_id(Some(prefix)), Some(first));
1330    }
1331
1332    #[rstest]
1333    fn venue_identity_can_be_bound_and_replaced() {
1334        let client_order_id = ClientOrderId::from("O-1");
1335        let mut state = OcmState::default();
1336        state
1337            .register_submission(client_order_id, StrategyId::from("S-001"))
1338            .unwrap();
1339
1340        state.bind_venue_order_id(&client_order_id, VenueOrderId::from("bet-1"));
1341        assert_eq!(
1342            state
1343                .order_correlations
1344                .get(&client_order_id)
1345                .and_then(|correlation| correlation.venue_order_id),
1346            Some(VenueOrderId::from("bet-1")),
1347        );
1348
1349        state.bind_venue_order_id(&client_order_id, VenueOrderId::from("bet-2"));
1350        assert_eq!(
1351            state
1352                .order_correlations
1353                .get(&client_order_id)
1354                .and_then(|correlation| correlation.venue_order_id),
1355            Some(VenueOrderId::from("bet-2")),
1356        );
1357    }
1358
1359    #[rstest]
1360    fn claimed_acceptance_does_not_replace_restored_venue_identity() {
1361        let client_order_id = ClientOrderId::from("O-1");
1362        let mut state = OcmState::default();
1363        state.restore_order(
1364            client_order_id,
1365            StrategyId::from("S-001"),
1366            VenueOrderId::from("bet-current"),
1367        );
1368
1369        let claimed = state.claim_acceptance(client_order_id, VenueOrderId::from("bet-stale"));
1370
1371        assert!(!claimed);
1372        assert_eq!(
1373            state
1374                .order_correlations
1375                .get(&client_order_id)
1376                .and_then(|correlation| correlation.venue_order_id),
1377            Some(VenueOrderId::from("bet-current")),
1378        );
1379    }
1380
1381    #[rstest]
1382    fn pending_replace_promotes_only_a_different_bet() {
1383        let client_order_id = ClientOrderId::from("O-1");
1384        let mut state = OcmState::default();
1385        state.restore_order(
1386            client_order_id,
1387            StrategyId::from("S-001"),
1388            VenueOrderId::from("old-bet"),
1389        );
1390        state.register_pending_replace(
1391            client_order_id,
1392            "old-bet".to_string(),
1393            Some(Quantity::from(10)),
1394        );
1395
1396        assert_eq!(
1397            state.promote_pending_replace(&client_order_id, "old-bet", Quantity::from(8)),
1398            None
1399        );
1400        assert_eq!(
1401            state.promote_pending_replace(&client_order_id, "new-bet", Quantity::from(8)),
1402            Some((Quantity::from(10), "old-bet".to_string())),
1403        );
1404        assert!(state.pending_replace_state.is_empty());
1405        assert!(state.replaced_venue_order_ids.contains("old-bet"));
1406        assert_eq!(
1407            state
1408                .order_correlations
1409                .get(&client_order_id)
1410                .and_then(|correlation| correlation.venue_order_id),
1411            Some(VenueOrderId::from("new-bet")),
1412        );
1413        assert_eq!(
1414            state.promote_pending_replace(&client_order_id, "newer-bet", Quantity::from(8)),
1415            None,
1416        );
1417    }
1418
1419    #[rstest]
1420    fn pending_replace_does_not_promote_a_historical_bet() {
1421        let client_order_id = ClientOrderId::from("O-1");
1422        let mut state = OcmState::default();
1423        state
1424            .replaced_venue_order_ids
1425            .insert("historical-bet".to_string());
1426        state.register_pending_replace(
1427            client_order_id,
1428            "current-bet".to_string(),
1429            Some(Quantity::from(10)),
1430        );
1431
1432        assert_eq!(
1433            state.promote_pending_replace(&client_order_id, "historical-bet", Quantity::from(8)),
1434            None,
1435        );
1436        assert!(
1437            state.should_suppress_cancel(&client_order_id, "current-bet"),
1438            "historical traffic must not consume the pending replace",
1439        );
1440        assert_eq!(
1441            state.promote_pending_replace(&client_order_id, "replacement-bet", Quantity::from(8)),
1442            Some((Quantity::from(10), "current-bet".to_string())),
1443        );
1444    }
1445
1446    #[rstest]
1447    #[case::below_requested(Quantity::from(3), None)]
1448    #[case::at_requested(Quantity::from(4), Some(Quantity::from(4)))]
1449    #[case::inside_window(Quantity::from(9), Some(Quantity::from(9)))]
1450    #[case::at_original(Quantity::from(10), None)]
1451    fn pending_reduction_confirms_only_a_definitive_reduction(
1452        #[case] active_quantity: Quantity,
1453        #[case] expected: Option<Quantity>,
1454    ) {
1455        let client_order_id = ClientOrderId::from("O-1");
1456        let bet_id = "bet-1";
1457        let mut state = OcmState::default();
1458        state.register_pending_reduction(
1459            client_order_id,
1460            bet_id.to_string(),
1461            Quantity::from(10),
1462            Quantity::from(4),
1463        );
1464
1465        assert_eq!(
1466            state.confirm_pending_reduction(&client_order_id, bet_id, active_quantity),
1467            expected,
1468        );
1469        assert_eq!(state.reduced_quantity(bet_id), expected);
1470    }
1471
1472    #[rstest]
1473    fn pending_reduction_validates_identity_and_resolves_once() {
1474        let client_order_id = ClientOrderId::from("O-1");
1475        let other_client_order_id = ClientOrderId::from("O-2");
1476        let bet_id = "bet-1";
1477        let mut state = OcmState::default();
1478        state.register_pending_reduction(
1479            client_order_id,
1480            bet_id.to_string(),
1481            Quantity::from(10),
1482            Quantity::from(4),
1483        );
1484
1485        let mismatched =
1486            state.confirm_pending_reduction(&other_client_order_id, bet_id, Quantity::from(4));
1487        assert_eq!(mismatched, None);
1488        assert!(!state.complete_pending_reduction(
1489            &other_client_order_id,
1490            bet_id,
1491            Quantity::from(4),
1492        ));
1493        state.clear_pending_reduction(&other_client_order_id, bet_id);
1494
1495        assert_eq!(
1496            state.confirm_pending_reduction(&client_order_id, bet_id, Quantity::from(4)),
1497            Some(Quantity::from(4)),
1498        );
1499        assert_eq!(
1500            state.confirm_pending_reduction(&client_order_id, bet_id, Quantity::from(4)),
1501            None,
1502            "a confirmed reduction must not resolve twice",
1503        );
1504
1505        state.clear_pending_reduction(&client_order_id, bet_id);
1506
1507        assert_eq!(state.reduced_quantity(bet_id), None);
1508        assert_eq!(
1509            state.confirm_pending_reduction(&client_order_id, bet_id, Quantity::from(4)),
1510            None,
1511            "a discarded reduction must not resolve from a later observation",
1512        );
1513    }
1514
1515    #[rstest]
1516    fn terminal_order_retention_evicts_pending_reduction_state() {
1517        let client_order_id = ClientOrderId::from("O-1");
1518        let bet_id = "bet-0";
1519        let mut state = OcmState::default();
1520        state.register_pending_reduction(
1521            client_order_id,
1522            bet_id.to_string(),
1523            Quantity::from(10),
1524            Quantity::from(4),
1525        );
1526        state.confirm_pending_reduction(&client_order_id, bet_id, Quantity::from(4));
1527
1528        for index in 0..=OcmState::DEDUP_RETENTION {
1529            state.mark_terminal_order(format!("bet-{index}"));
1530        }
1531
1532        assert_eq!(state.reduced_quantity(bet_id), None);
1533    }
1534
1535    #[rstest]
1536    fn canceled_replace_keeps_one_terminal_retention_entry() {
1537        let mut state = OcmState::default();
1538        let client_order_id = ClientOrderId::from("O-1");
1539
1540        state.mark_canceled_replace(client_order_id, "old-bet");
1541        state.retain_terminal_order(client_order_id, "old-bet");
1542
1543        assert!(state.terminal_orders.contains("old-bet"));
1544        assert_eq!(state.terminal_order_queue.len(), 1);
1545    }
1546
1547    #[rstest]
1548    fn canceled_replace_without_ocm_has_terminal_retention_entry() {
1549        let mut state = OcmState::default();
1550        let client_order_id = ClientOrderId::from("O-1");
1551
1552        state.mark_canceled_replace(client_order_id, "old-bet");
1553
1554        assert!(state.terminal_orders.contains("old-bet"));
1555        assert_eq!(state.terminal_order_queue.len(), 1);
1556    }
1557}