futu_gateway_core/bridge/
broker_runtime.rs

1//! Broker/backend connection runtime boundary for [`super::GatewayBridge`].
2//!
3//! This bundle owns the shared Platform backend pointer and broker-channel
4//! connection maps. `GatewayBridge` exposes this bundle directly instead of
5//! publishing separate connection alias fields, so startup, push, dispatcher,
6//! and handler registration paths share one narrow owner boundary.
7
8use std::collections::BTreeSet;
9use std::fmt::Write as _;
10use std::sync::Arc;
11
12use arc_swap::ArcSwap;
13use futu_backend::auth::BrokerAuth;
14use futu_backend::conn::BackendConn;
15use futu_backend::overnight_whitelist::OvernightWhitelistCache;
16
17use super::clock::wall_clock_since_unix_epoch_or_zero;
18use super::types::BrokerStatus;
19use super::{BrokerConnMeta, SharedBackend};
20
21pub const BROKER_RECONNECT_MAX_CONSECUTIVE_FAILURES: u32 = 6;
22
23#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
24enum BrokerChannelState {
25    Connected,
26    SustainDisconnected,
27    #[default]
28    Reconnecting,
29}
30
31impl BrokerChannelState {
32    fn status_label(self, backend_present: bool, connected: bool) -> &'static str {
33        match self {
34            Self::Connected if backend_present && connected => "ready",
35            Self::Connected if backend_present => "disconnected",
36            Self::Connected | Self::Reconnecting => "reconnecting",
37            Self::SustainDisconnected => "sustain_disconnected",
38        }
39    }
40}
41
42#[derive(Debug, Clone, Default)]
43struct BrokerHealthState {
44    channel_state: BrokerChannelState,
45    consecutive_failures: u32,
46    max_consecutive_failures: u32,
47    last_reconnect_attempt_at_ms: i64,
48    last_reconnect_success_at_ms: i64,
49    last_reconnect_failure_at_ms: i64,
50}
51
52#[derive(Clone)]
53pub struct BrokerRuntime {
54    backend: SharedBackend,
55    brokers: Arc<dashmap::DashMap<u32, Arc<BackendConn>>>,
56    broker_conn_meta: Arc<dashmap::DashMap<u32, BrokerConnMeta>>,
57    broker_health: Arc<dashmap::DashMap<u32, BrokerHealthState>>,
58    overnight_whitelist: OvernightWhitelistCache,
59    overnight_whitelist_workers: Arc<dashmap::DashSet<u32>>,
60}
61
62impl BrokerRuntime {
63    pub fn empty() -> Self {
64        let backend: SharedBackend = Arc::new(ArcSwap::new(Arc::new(None)));
65        Self::new(
66            backend,
67            Arc::new(dashmap::DashMap::new()),
68            Arc::new(dashmap::DashMap::new()),
69        )
70    }
71
72    pub fn new(
73        backend: SharedBackend,
74        brokers: Arc<dashmap::DashMap<u32, Arc<BackendConn>>>,
75        broker_conn_meta: Arc<dashmap::DashMap<u32, BrokerConnMeta>>,
76    ) -> Self {
77        Self {
78            backend,
79            brokers,
80            broker_conn_meta,
81            broker_health: Arc::new(dashmap::DashMap::new()),
82            overnight_whitelist: OvernightWhitelistCache::new(),
83            overnight_whitelist_workers: Arc::new(dashmap::DashSet::new()),
84        }
85    }
86
87    pub fn platform_backend(&self) -> Option<Arc<BackendConn>> {
88        self.backend.load().as_ref().clone()
89    }
90
91    pub fn set_platform_backend(&self, backend: Arc<BackendConn>) {
92        self.backend.store(Arc::new(Some(backend)));
93    }
94
95    /// Shared CMD20874 overnight whitelist cache.
96    ///
97    /// Owner boundary: broker startup/reconnect refreshes this cache; trade
98    /// write handlers only read it while computing C++ `AdjustTradeSession`.
99    pub fn overnight_whitelist(&self) -> OvernightWhitelistCache {
100        self.overnight_whitelist.clone()
101    }
102
103    pub fn mark_overnight_whitelist_worker_started(&self, broker_id: u32) -> bool {
104        self.overnight_whitelist_workers.insert(broker_id)
105    }
106
107    pub fn platform_connected(&self) -> bool {
108        self.platform_backend()
109            .as_ref()
110            .is_some_and(|backend| backend.is_connected())
111    }
112
113    pub fn clear_platform_backend(&self) {
114        self.backend.store(Arc::new(None));
115    }
116
117    pub fn broker_backend(&self, broker_id: u32) -> Option<Arc<BackendConn>> {
118        self.brokers.get(&broker_id).and_then(|conn| {
119            if conn.is_connected() {
120                Some(Arc::clone(conn.value()))
121            } else {
122                None
123            }
124        })
125    }
126
127    pub fn broker_customer_id(&self, broker_id: u32) -> Option<u64> {
128        self.broker_conn_meta
129            .get(&broker_id)
130            .map(|meta| meta.broker_auth.customer_id)
131    }
132
133    pub fn broker_meta(&self, broker_id: u32) -> Option<BrokerConnMeta> {
134        self.broker_conn_meta
135            .get(&broker_id)
136            .map(|meta| meta.value().clone())
137    }
138
139    pub fn active_broker_ids(&self) -> Vec<u32> {
140        self.brokers.iter().map(|entry| *entry.key()).collect()
141    }
142
143    pub fn all_broker_backends(&self) -> Vec<(u32, Arc<BackendConn>)> {
144        let mut out: Vec<_> = self
145            .brokers
146            .iter()
147            .filter_map(|r| {
148                if r.value().is_connected() {
149                    Some((*r.key(), Arc::clone(r.value())))
150                } else {
151                    None
152                }
153            })
154            .collect();
155        out.sort_by_key(|(id, _)| *id);
156        out
157    }
158
159    /// Mark every broker backend disconnected without deleting its meta.
160    ///
161    /// Platform reconnect can invalidate broker-side session semantics while
162    /// the broker TCP socket still appears open. Failing closed here prevents
163    /// CMD2900 / trade queries from using a stale broker session; the existing
164    /// per-broker heartbeat watcher observes the disconnected backend and
165    /// reconnects with the cached broker auth.
166    pub fn mark_all_broker_backends_disconnected(&self, reason: &'static str) -> usize {
167        let mut count = 0;
168        for entry in self.brokers.iter() {
169            self.mark_broker_sustain_disconnected(*entry.key(), reason);
170            entry.value().mark_disconnected_for_reconnect(reason);
171            count += 1;
172        }
173        if count > 0 {
174            tracing::warn!(
175                broker_count = count,
176                reason,
177                "marked broker backends disconnected for reconnect"
178            );
179        }
180        count
181    }
182
183    pub fn remove_broker_channel(&self, broker_id: u32) {
184        self.brokers.remove(&broker_id);
185        self.broker_conn_meta.remove(&broker_id);
186        self.broker_health.remove(&broker_id);
187    }
188
189    pub fn remove_broker_backend(&self, broker_id: u32) {
190        self.brokers.remove(&broker_id);
191    }
192
193    pub fn mark_broker_sustain_disconnected(&self, broker_id: u32, reason: &'static str) {
194        self.update_broker_health(broker_id, |state| {
195            state.channel_state = BrokerChannelState::SustainDisconnected;
196            if state.max_consecutive_failures == 0 {
197                state.max_consecutive_failures = BROKER_RECONNECT_MAX_CONSECUTIVE_FAILURES;
198            }
199        });
200        tracing::warn!(
201            broker_id,
202            reason,
203            "broker channel entered C++ sustain-disconnected state"
204        );
205    }
206
207    pub fn update_broker_auth(&self, broker_id: u32, broker_auth: BrokerAuth) -> bool {
208        if let Some(mut entry) = self.broker_conn_meta.get_mut(&broker_id) {
209            entry.broker_auth = broker_auth;
210            true
211        } else {
212            false
213        }
214    }
215
216    pub fn register_broker_channel(
217        &self,
218        broker_id: u32,
219        conn: Arc<BackendConn>,
220        meta: BrokerConnMeta,
221    ) {
222        self.broker_conn_meta.insert(broker_id, meta);
223        self.brokers.insert(broker_id, conn);
224        self.mark_broker_reconnect_success(broker_id);
225    }
226
227    pub fn set_broker_meta(&self, broker_id: u32, meta: BrokerConnMeta) {
228        self.broker_conn_meta.insert(broker_id, meta);
229    }
230
231    pub fn set_broker_backend(&self, broker_id: u32, conn: Arc<BackendConn>) {
232        self.brokers.insert(broker_id, conn);
233        self.mark_broker_reconnect_success(broker_id);
234    }
235
236    pub fn mark_broker_reconnecting(&self, broker_id: u32, max_failures: u32) {
237        let now_ms = unix_now_ms("broker_reconnect_attempt");
238        self.update_broker_health(broker_id, |state| {
239            state.channel_state = BrokerChannelState::Reconnecting;
240            state.max_consecutive_failures = max_failures;
241            state.last_reconnect_attempt_at_ms = now_ms;
242        });
243    }
244
245    pub fn mark_broker_reconnect_attempt(&self, broker_id: u32, max_failures: u32) {
246        self.mark_broker_reconnecting(broker_id, max_failures);
247    }
248
249    pub fn mark_broker_reconnect_failure(
250        &self,
251        broker_id: u32,
252        consecutive_failures: u32,
253        max_failures: u32,
254    ) {
255        let now_ms = unix_now_ms("broker_reconnect_failure");
256        self.update_broker_health(broker_id, |state| {
257            state.channel_state = BrokerChannelState::Reconnecting;
258            state.consecutive_failures = consecutive_failures;
259            state.max_consecutive_failures = max_failures;
260            state.last_reconnect_failure_at_ms = now_ms;
261        });
262    }
263
264    pub fn mark_broker_reconnect_success(&self, broker_id: u32) {
265        let now_ms = unix_now_ms("broker_reconnect_success");
266        self.update_broker_health(broker_id, |state| {
267            state.channel_state = BrokerChannelState::Connected;
268            state.consecutive_failures = 0;
269            if state.max_consecutive_failures == 0 {
270                state.max_consecutive_failures = BROKER_RECONNECT_MAX_CONSECUTIVE_FAILURES;
271            }
272            state.last_reconnect_success_at_ms = now_ms;
273        });
274    }
275
276    fn update_broker_health(&self, broker_id: u32, update: impl FnOnce(&mut BrokerHealthState)) {
277        let mut entry = self.broker_health.entry(broker_id).or_default();
278        update(&mut entry);
279    }
280
281    pub fn broker_statuses(&self) -> Vec<BrokerStatus> {
282        let mut ids = BTreeSet::new();
283        ids.extend(self.brokers.iter().map(|entry| *entry.key()));
284        ids.extend(self.broker_conn_meta.iter().map(|entry| *entry.key()));
285        ids.extend(self.broker_health.iter().map(|entry| *entry.key()));
286
287        ids.into_iter()
288            .map(|broker_id| {
289                let name = futu_backend::auth::broker_config(broker_id)
290                    .map(|c| c.name)
291                    .unwrap_or("unknown");
292                let meta = self.broker_conn_meta.get(&broker_id);
293                let backend = self.broker_backend(broker_id);
294                let connected = backend.as_ref().is_some_and(|conn| conn.is_connected());
295                let health = self
296                    .broker_health
297                    .get(&broker_id)
298                    .map(|entry| entry.value().clone())
299                    .unwrap_or_default();
300                let backend_present = self.brokers.contains_key(&broker_id);
301                let status = health
302                    .channel_state
303                    .status_label(backend_present, connected)
304                    .to_string();
305
306                BrokerStatus {
307                    broker_id,
308                    name: name.to_string(),
309                    status,
310                    connected,
311                    addr: meta.as_ref().map(|m| m.connect_addr.clone()),
312                    keep_alive_secs: meta.as_ref().map(|m| m.keep_alive_secs),
313                    consecutive_failures: health.consecutive_failures,
314                    max_consecutive_failures: health.max_consecutive_failures,
315                    last_reconnect_attempt_at_ms: non_zero_ms(health.last_reconnect_attempt_at_ms),
316                    last_reconnect_success_at_ms: non_zero_ms(health.last_reconnect_success_at_ms),
317                    last_reconnect_failure_at_ms: non_zero_ms(health.last_reconnect_failure_at_ms),
318                }
319            })
320            .collect()
321    }
322
323    /// Render per-broker TCP health as Prometheus text exposition.
324    ///
325    /// This intentionally reuses [`BrokerRuntime::broker_statuses`] as the
326    /// single status source. Metrics labels avoid `addr` because broker TCP
327    /// endpoints are dynamic infrastructure details and high-cardinality.
328    #[must_use]
329    pub fn render_prometheus(&self) -> String {
330        let statuses = self.broker_statuses();
331        let mut s = String::with_capacity(1024 + statuses.len() * 512);
332
333        s.push_str(
334            "# HELP futu_gateway_broker_tcp_connected Broker TCP connected state (1=connected,0=not connected)\n# TYPE futu_gateway_broker_tcp_connected gauge\n",
335        );
336        for status in &statuses {
337            let labels = broker_metric_labels(status);
338            let _ = writeln!(
339                s,
340                "futu_gateway_broker_tcp_connected{{{labels}}} {}",
341                u8::from(status.connected)
342            );
343        }
344
345        s.push_str(
346            "# HELP futu_gateway_broker_tcp_reconnect_consecutive_failures Consecutive broker TCP reconnect failures\n# TYPE futu_gateway_broker_tcp_reconnect_consecutive_failures gauge\n",
347        );
348        for status in &statuses {
349            let labels = broker_metric_labels(status);
350            let _ = writeln!(
351                s,
352                "futu_gateway_broker_tcp_reconnect_consecutive_failures{{{labels}}} {}",
353                status.consecutive_failures
354            );
355        }
356
357        s.push_str(
358            "# HELP futu_gateway_broker_tcp_reconnect_max_consecutive_failures Broker TCP reconnect failure count before RepullAuthCode self-heal\n# TYPE futu_gateway_broker_tcp_reconnect_max_consecutive_failures gauge\n",
359        );
360        for status in &statuses {
361            let labels = broker_metric_labels(status);
362            let _ = writeln!(
363                s,
364                "futu_gateway_broker_tcp_reconnect_max_consecutive_failures{{{labels}}} {}",
365                status.max_consecutive_failures
366            );
367        }
368
369        s.push_str(
370            "# HELP futu_gateway_broker_tcp_reconnect_last_attempt_at_ms Last broker TCP reconnect attempt timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_broker_tcp_reconnect_last_attempt_at_ms gauge\n",
371        );
372        for status in &statuses {
373            let labels = broker_metric_labels(status);
374            let _ = writeln!(
375                s,
376                "futu_gateway_broker_tcp_reconnect_last_attempt_at_ms{{{labels}}} {}",
377                status.last_reconnect_attempt_at_ms.unwrap_or(0)
378            );
379        }
380
381        s.push_str(
382            "# HELP futu_gateway_broker_tcp_reconnect_last_success_at_ms Last broker TCP reconnect success timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_broker_tcp_reconnect_last_success_at_ms gauge\n",
383        );
384        for status in &statuses {
385            let labels = broker_metric_labels(status);
386            let _ = writeln!(
387                s,
388                "futu_gateway_broker_tcp_reconnect_last_success_at_ms{{{labels}}} {}",
389                status.last_reconnect_success_at_ms.unwrap_or(0)
390            );
391        }
392
393        s.push_str(
394            "# HELP futu_gateway_broker_tcp_reconnect_last_failure_at_ms Last broker TCP reconnect failure timestamp in Unix milliseconds (0=never)\n# TYPE futu_gateway_broker_tcp_reconnect_last_failure_at_ms gauge\n",
395        );
396        for status in &statuses {
397            let labels = broker_metric_labels(status);
398            let _ = writeln!(
399                s,
400                "futu_gateway_broker_tcp_reconnect_last_failure_at_ms{{{labels}}} {}",
401                status.last_reconnect_failure_at_ms.unwrap_or(0)
402            );
403        }
404
405        s
406    }
407}
408
409fn non_zero_ms(value: i64) -> Option<i64> {
410    (value > 0).then_some(value)
411}
412
413fn unix_now_ms(context: &'static str) -> i64 {
414    let elapsed = wall_clock_since_unix_epoch_or_zero(context).as_millis();
415    elapsed.min(i64::MAX as u128) as i64
416}
417
418fn broker_metric_labels(status: &BrokerStatus) -> String {
419    format!(
420        "broker_id=\"{}\",broker={}",
421        status.broker_id,
422        prom_label(&status.name)
423    )
424}
425
426fn prom_label(value: &str) -> String {
427    let mut out = String::with_capacity(value.len() + 2);
428    out.push('"');
429    for c in value.chars() {
430        match c {
431            '\\' => out.push_str("\\\\"),
432            '"' => out.push_str("\\\""),
433            '\n' => out.push_str("\\n"),
434            _ => out.push(c),
435        }
436    }
437    out.push('"');
438    out
439}
440
441impl Default for BrokerRuntime {
442    fn default() -> Self {
443        Self::empty()
444    }
445}
446
447#[cfg(test)]
448mod tests;
futu_gateway_core/bridge/broker_runtime.rs

futu_gateway_core/bridge/
broker_runtime.rs
