futu_cache/trd_cache/
order_relation_snapshot.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::sync::Arc;
3
4use super::{AccKey, CachedOrder, TrdCache};
5
6#[derive(Debug, Clone, Copy, PartialEq, Eq)]
7pub enum OrderRelationSnapshotAvailability {
8    Missing,
9    Fresh,
10    Stale,
11}
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
14pub enum OrderRelationSourceChannel {
15    Unknown,
16    Platform,
17    Broker(u32),
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub struct OrderRelationSourceToken {
22    pub channel: OrderRelationSourceChannel,
23    pub connection_generation: u64,
24}
25
26impl OrderRelationSourceToken {
27    #[must_use]
28    pub const fn unknown(connection_generation: u64) -> Self {
29        Self {
30            channel: OrderRelationSourceChannel::Unknown,
31            connection_generation,
32        }
33    }
34
35    #[must_use]
36    pub const fn platform(connection_generation: u64) -> Self {
37        Self {
38            channel: OrderRelationSourceChannel::Platform,
39            connection_generation,
40        }
41    }
42
43    #[must_use]
44    pub const fn broker(broker_id: u32, connection_generation: u64) -> Self {
45        Self {
46            channel: OrderRelationSourceChannel::Broker(broker_id),
47            connection_generation,
48        }
49    }
50}
51
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub(super) struct OrderRelationSnapshotState {
54    pub(super) availability: OrderRelationSnapshotAvailability,
55    pub(super) sequence: u64,
56    /// A Platform-wide reconnect invalidates every previously created source.
57    pub(super) minimum_global_connection_generation: u64,
58    /// A broker-only reconnect rejects that exact retired connection without
59    /// ordering unrelated live broker/platform sources against each other.
60    pub(super) source_minimum_generations: BTreeMap<OrderRelationSourceChannel, u64>,
61}
62
63impl Default for OrderRelationSnapshotState {
64    fn default() -> Self {
65        Self {
66            availability: OrderRelationSnapshotAvailability::Missing,
67            sequence: 0,
68            minimum_global_connection_generation: 0,
69            source_minimum_generations: BTreeMap::new(),
70        }
71    }
72}
73
74#[derive(Debug, Clone)]
75pub struct OrderRelationSnapshotLookup {
76    pub availability: OrderRelationSnapshotAvailability,
77    pub sequence: u64,
78    pub orders: Vec<CachedOrder>,
79}
80
81impl TrdCache {
82    pub(super) fn order_relation_snapshot_lock(
83        &self,
84        acc_id: AccKey,
85    ) -> Arc<parking_lot::Mutex<()>> {
86        self.order_relation_snapshot_locks
87            .entry(acc_id)
88            .or_insert_with(|| Arc::new(parking_lot::Mutex::new(())))
89            .clone()
90    }
91
92    /// Read one immutable, completeness-qualified order snapshot. The 22715
93    /// handler only calls this method and never performs backend I/O.
94    #[must_use]
95    pub fn order_relation_snapshot(&self, acc_id: AccKey) -> OrderRelationSnapshotLookup {
96        let relation_lock = self.order_relation_snapshot_lock(acc_id);
97        let _relation_guard = relation_lock.lock();
98        let availability = self
99            .order_relation_snapshot_states
100            .get(&acc_id)
101            .map(|value| value.availability)
102            .unwrap_or(OrderRelationSnapshotAvailability::Missing);
103        let orders = self
104            .orders
105            .get(&acc_id)
106            .map(|orders| orders.clone())
107            .unwrap_or_default();
108        OrderRelationSnapshotLookup {
109            availability,
110            sequence: self.order_relation_snapshot_state(acc_id).sequence,
111            orders,
112        }
113    }
114
115    #[must_use]
116    pub fn order_relation_snapshot_is_current(&self, acc_id: AccKey, sequence: u64) -> bool {
117        let relation_lock = self.order_relation_snapshot_lock(acc_id);
118        let _relation_guard = relation_lock.lock();
119        let state = self.order_relation_snapshot_state(acc_id);
120        state.availability == OrderRelationSnapshotAvailability::Fresh && state.sequence == sequence
121    }
122
123    /// Fence every account to the newly published Platform generation. Older
124    /// in-flight responses can no longer turn Stale back into Fresh.
125    pub fn invalidate_all_order_relation_snapshots_for_reconnect(
126        &self,
127        minimum_global_connection_generation: u64,
128    ) {
129        let mut acc_ids = self
130            .accounts
131            .iter()
132            .map(|entry| *entry.key())
133            .collect::<BTreeSet<_>>();
134        acc_ids.extend(
135            self.order_relation_snapshot_states
136                .iter()
137                .map(|entry| *entry.key()),
138        );
139        for acc_id in acc_ids {
140            self.invalidate_order_relation_snapshot_all_sources(
141                acc_id,
142                minimum_global_connection_generation,
143            );
144        }
145    }
146
147    /// Invalidate accounts routed through the reconnecting broker and accounts
148    /// carrying an explicit order->broker route to it.
149    pub fn invalidate_order_relation_snapshots_for_broker_reconnect(
150        &self,
151        broker_id: u32,
152        rejected_connection_generation: Option<u64>,
153    ) {
154        let mut acc_ids = self
155            .accounts
156            .iter()
157            .filter_map(|entry| {
158                let account = entry.value();
159                let account_broker = account
160                    .security_firm
161                    .and_then(futu_core::trade_broker::broker_id_for_security_firm_like_cpp)
162                    .or_else(|| u32::try_from(account.sort_key >> 48).ok())?;
163                (account_broker == broker_id).then_some(account.acc_id)
164            })
165            .collect::<BTreeSet<_>>();
166        for entry in &self.order_broker_ids_by_acc {
167            if entry.value().iter().any(|order_id| {
168                self.order_brokers
169                    .get(order_id)
170                    .is_some_and(|mapped| *mapped == broker_id)
171            }) {
172                acc_ids.insert(*entry.key());
173            }
174        }
175        for acc_id in acc_ids {
176            self.invalidate_order_relation_snapshot_broker_source(
177                acc_id,
178                broker_id,
179                rejected_connection_generation,
180            );
181        }
182    }
183
184    pub(super) fn invalidate_order_relation_snapshot_all_sources(
185        &self,
186        acc_id: AccKey,
187        minimum_global_connection_generation: u64,
188    ) {
189        let relation_lock = self.order_relation_snapshot_lock(acc_id);
190        let _relation_guard = relation_lock.lock();
191        let mut state = self.order_relation_snapshot_state(acc_id);
192        state.availability = OrderRelationSnapshotAvailability::Stale;
193        state.sequence = self.take_order_relation_sequence();
194        state.minimum_global_connection_generation = state
195            .minimum_global_connection_generation
196            .max(minimum_global_connection_generation);
197        state.source_minimum_generations.clear();
198        self.order_relation_snapshot_states.insert(acc_id, state);
199    }
200
201    pub(super) fn invalidate_order_relation_snapshot_broker_source(
202        &self,
203        acc_id: AccKey,
204        broker_id: u32,
205        retired_connection_generation: Option<u64>,
206    ) {
207        let relation_lock = self.order_relation_snapshot_lock(acc_id);
208        let _relation_guard = relation_lock.lock();
209        let mut state = self.order_relation_snapshot_state(acc_id);
210        state.availability = OrderRelationSnapshotAvailability::Stale;
211        state.sequence = self.take_order_relation_sequence();
212        if let Some(retired) = retired_connection_generation {
213            let minimum = retired.saturating_add(1);
214            state
215                .source_minimum_generations
216                .entry(OrderRelationSourceChannel::Broker(broker_id))
217                .and_modify(|existing| *existing = (*existing).max(minimum))
218                .or_insert(minimum);
219        }
220        self.order_relation_snapshot_states.insert(acc_id, state);
221    }
222
223    /// Record a dropped/malformed row from an admitted source. A fenced older
224    /// source cannot mutate the current completeness state.
225    pub fn mark_order_relation_snapshot_incomplete_from_source(
226        &self,
227        acc_id: AccKey,
228        source: OrderRelationSourceToken,
229    ) {
230        let relation_lock = self.order_relation_snapshot_lock(acc_id);
231        let _relation_guard = relation_lock.lock();
232        let mut state = self.order_relation_snapshot_state(acc_id);
233        if !Self::order_relation_source_is_admitted(&state, source) {
234            return;
235        }
236        state.availability = OrderRelationSnapshotAvailability::Stale;
237        state.sequence = self.take_order_relation_sequence();
238        self.order_relation_snapshot_states.insert(acc_id, state);
239    }
240
241    pub(super) fn order_relation_snapshot_state(
242        &self,
243        acc_id: AccKey,
244    ) -> OrderRelationSnapshotState {
245        self.order_relation_snapshot_states
246            .get(&acc_id)
247            .map(|value| value.clone())
248            .unwrap_or_default()
249    }
250
251    pub(super) fn order_relation_source_is_admitted(
252        state: &OrderRelationSnapshotState,
253        source: OrderRelationSourceToken,
254    ) -> bool {
255        source.connection_generation >= state.minimum_global_connection_generation
256            && source.connection_generation
257                >= state
258                    .source_minimum_generations
259                    .get(&source.channel)
260                    .copied()
261                    .unwrap_or(0)
262    }
263
264    pub(super) fn publish_order_relation_snapshot_state(
265        &self,
266        acc_id: AccKey,
267        availability: OrderRelationSnapshotAvailability,
268        _source: Option<OrderRelationSourceToken>,
269    ) {
270        let mut state = self.order_relation_snapshot_state(acc_id);
271        state.availability = availability;
272        state.sequence = self.take_order_relation_sequence();
273        self.order_relation_snapshot_states.insert(acc_id, state);
274    }
275
276    fn take_order_relation_sequence(&self) -> u64 {
277        self.order_relation_next_sequence
278            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
279    }
280}
futu_cache/trd_cache/order_relation_snapshot.rs

futu_cache/trd_cache/
order_relation_snapshot.rs
