futu_gateway_core/bridge/push_health/
push.rs

1//! bridge/push_health/push — PushHealth + SharedPushHealth + helpers
2//! (v1.4.110 CC Batch P: 拆自 push_health.rs L1-654)
3
4//! v1.4.83 §9 Phase 2: Push channel health tracking (F2-F6 chain recovery foundation)
5//!
6//! **背景**：双 tester v1.4.81 §9 报告 daemon 对 push 通道健康完全 blind：
7//! - push parse 失败 silent log，无 retry / 无计数
8//! - push 断流 (>N 秒没收到) 无感知，`/api/push-subscriber-info` 撒谎返 "OK"
9//! - 连错 5+ 不 re-subscribe、连错 20+ 不 circuit breaker
10//! - 订单 stub status (0/1/2/4 in-flight) 永远卡住没 orphan 检测
11//!
12//! **F2-F6 chain recovery 架构**：
13//! - **F2** (retry): push parse 无意义（frame 已消耗），改做 **TradeReQuery
14//!   backend call 的 exponential backoff** (1s/3s/9s)
15//! - **F3** (re-subscribe): `consecutive_parse_errors >= 5` 或 **3-in-2min**
16//!   触发 resubscribe_quotes
17//! - **F4** (circuit breaker): `consecutive_parse_errors >= 20` → 设
18//!   `tripped=true`，30s cooldown 后自动 reset
19//! - **F5** (expose): `/api/push-subscriber-info` 返真实 `push_stream_healthy`
20//!   / `last_push_received_at_ms` / `consecutive_parse_errors` 等字段
21//! - **F6** (orphan detect): tokio 30s interval task 扫 `trd_cache` 查
22//!   `order_status ∈ {0,1,2,4} (in-flight stub) && now - create_timestamp > 300s`
23//!   → warn log + metric. v1.4.103 BUG-WUZONG-001 把 status=0 (Unsubmitted)
24//!   也 enroll, 防 daemon-side stub 永远 stale.
25//!
26//! **同步约束**: push callback 是 Arc<dyn Fn(u32, Vec<u8>) + Send + Sync>
27//! 同步 closure (见 `futu_backend::conn::PushCallback`), 所以 `PushHealth`
28//! 用 atomic 无锁更新 + Arc 共享跨线程.
29
30use std::collections::HashMap;
31use std::sync::Arc;
32use std::sync::Mutex;
33use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, AtomicU64, Ordering};
34
35mod f2;
36
37use super::snapshot::{PushHealthSnapshot, PushSubTypeWatermark};
38use f2::F2AccState;
39#[cfg(test)]
40pub(super) use f2::{F2_COOLDOWN_MS, F2_TRIP_THRESHOLD};
41
42/// v1.4.92 P1-D Tier A2: 连续失败 → 触发 circuit_open 的阈值.
43pub const RELOGIN_CIRCUIT_TRIP_THRESHOLD: u32 = 5;
44/// v1.4.92 P1-D Tier A2: circuit_open 持续时长 — 期间跳过所有 attempt
45/// (10 min, 与 P1-D failure backoff 600s cap 对齐).
46pub const RELOGIN_CIRCUIT_COOLDOWN_MS: i64 = 600_000;
47
48/// v1.4.83 §9 F2-F6 推送通道健康追踪。
49///
50/// **所有字段 atomic 无锁**，安全跨 sync push callback / async dispatcher /
51/// async orphan scan task / REST handler 多个执行上下文。
52///
53/// **关键约束**：
54/// - `last_push_received_at_ms`：自 daemon 启动后**任意** push 收到就更新；
55///   用作 staleness watermark（`/api/push-subscriber-info` F5 暴露）
56/// - `consecutive_parse_errors`：成功 push 清零；失败 += 1；触发 F3/F4
57/// - `total_parse_errors`：monotonic，用于 metrics rate 计算
58/// - `circuit_breaker_tripped_ms`：F4 tripped 时写入 Unix ms；dispatcher
59///   在 `now - tripped_ms < 30_000` 时跳过 dispatch（30s cooldown）
60#[derive(Debug)]
61pub struct PushHealth {
62    /// 最近一次成功 push 的 Unix 毫秒时间戳（0 = 还没收到）
63    last_push_received_at_ms: AtomicI64,
64    /// 最近一次按 FTAPI SubType 成功分发的 QOT push Unix ms.
65    ///
66    /// 全局 `last_push_received_at_ms` 只能证明 backend push TCP 活着；混合订阅
67    /// 场景里 QUOTE 持续更新会掩盖 ORDER_BOOK/KL 等单类型断流。这里按
68    /// sub_type 记录 dispatcher 已收到的普通 QuotePush，供 REST/Prometheus 排障。
69    last_push_by_sub_type: Mutex<HashMap<i32, i64>>,
70    /// 连续 parse 错误计数（成功则清零）
71    consecutive_parse_errors: AtomicU32,
72    /// 总累积 parse 错误（monotonic）
73    total_parse_errors: AtomicU64,
74    /// 成功 push 总数（monotonic，同 metrics.backend_pushes_received 但独立读）
75    total_pushes_received: AtomicU64,
76    /// F3 触发 re-subscribe 的次数
77    resubscribe_triggers: AtomicU64,
78    /// F4 circuit breaker tripped 时间（Unix ms，0 = 未 tripped）
79    circuit_breaker_tripped_at_ms: AtomicI64,
80    /// F4 circuit breaker tripped 总次数
81    circuit_breaker_trips: AtomicU64,
82    /// F6 orphan order 探测次数（每次 scan 发现 count 即 ++）
83    orphan_orders_detected: AtomicU64,
84    /// 最近一次 orphan scan 的 Unix ms（给 F5 暴露）
85    last_orphan_scan_at_ms: AtomicI64,
86    /// 最近一次 re-subscribe 的 Unix ms（给 F5 暴露）
87    last_resubscribe_at_ms: AtomicI64,
88    /// F4 is-tripped-now 快速 check（不用每次 load + 计算 cooldown）
89    is_circuit_tripped: AtomicBool,
90
91    /// v1.4.90 P2-B: F2 (TradeReQuery) account-level circuit-break 总 trip 次数.
92    /// Prometheus 暴露, 用于 alert 频繁 F2 失败的 account.
93    f2_circuit_tripped_total: AtomicU64,
94
95    /// v1.4.90 P2-B: 每 account F2 状态 (exhaust 窗口 + trip 时间).
96    /// Mutex 锁串行, 大部分 acc 没频繁 F2, 锁竞争极低.
97    f2_acc_state: Mutex<HashMap<u32, F2AccState>>,
98
99    /// v1.4.106 codex 0932 F1: TradeReQuery 队列满 / closed 时丢弃总数.
100    /// 旧实装 `let _ = push_tx.try_send(...)` 把 `TrySendError::Full /
101    /// Closed` silently 吞掉, push_callback 仍 record_push_success() →
102    /// downstream cache 永远 stale. 本 counter 暴露给 Prometheus +
103    /// `/api/push-subscriber-info`, 让 ops 能发现 backpressure.
104    ///
105    /// 同时, 队列满 = backpressure 信号; 持续命中 = degraded → next snapshot
106    /// 报 unhealthy 让上层 (REST `/healthz` / SubscriberInfo F5) 看到.
107    trade_requery_queue_dropped_total: AtomicU64,
108
109    /// v1.4.106 codex 0932 F1: 最近一次 TradeReQuery 队列丢弃的 Unix ms.
110    /// 0 = 从未丢弃过. 用于 health degraded 判定: dropped 后 60s 内
111    /// `is_healthy()` 应返 false (让 staleness detector / dispatcher 看到
112    /// degraded).
113    last_trade_requery_drop_at_ms: AtomicI64,
114
115    /// **v1.4.106 codex 0631 F4 [P2]**: 最近一次 resubscribe 部分失败的 Unix ms.
116    /// 0 = 从未部分失败 (健康). dispatcher 根据本字段判 push stream 健康度
117    /// (60s 内 partial → degraded).
118    last_resubscribe_partial_at_ms: AtomicI64,
119    /// **v1.4.106 codex 0631 F4 [P2]**: resubscribe 部分失败累计次数.
120    /// 监控 alarm 用. 与 `resubscribe_triggers` 区分 — 后者是触发次数,
121    /// 本字段是部分失败次数.
122    resubscribe_partial_total: AtomicU64,
123    /// **v1.4.106 codex 0631 F4 [P2]**: 最近一次 resubscribe 缺失 (cache miss /
124    /// unresolved) 的 keys 数. 0 = 全部 resolve OK.
125    last_missed_keys_count: AtomicU32,
126}
127
128impl Default for PushHealth {
129    fn default() -> Self {
130        Self::new()
131    }
132}
133
134impl PushHealth {
135    pub fn new() -> Self {
136        Self {
137            last_push_received_at_ms: AtomicI64::new(0),
138            last_push_by_sub_type: Mutex::new(HashMap::new()),
139            consecutive_parse_errors: AtomicU32::new(0),
140            total_parse_errors: AtomicU64::new(0),
141            total_pushes_received: AtomicU64::new(0),
142            resubscribe_triggers: AtomicU64::new(0),
143            circuit_breaker_tripped_at_ms: AtomicI64::new(0),
144            circuit_breaker_trips: AtomicU64::new(0),
145            orphan_orders_detected: AtomicU64::new(0),
146            last_orphan_scan_at_ms: AtomicI64::new(0),
147            last_resubscribe_at_ms: AtomicI64::new(0),
148            is_circuit_tripped: AtomicBool::new(false),
149            f2_circuit_tripped_total: AtomicU64::new(0),
150            f2_acc_state: Mutex::new(HashMap::new()),
151            trade_requery_queue_dropped_total: AtomicU64::new(0),
152            last_trade_requery_drop_at_ms: AtomicI64::new(0),
153            last_resubscribe_partial_at_ms: AtomicI64::new(0),
154            resubscribe_partial_total: AtomicU64::new(0),
155            last_missed_keys_count: AtomicU32::new(0),
156        }
157    }
158
159    /// 调用时机：每次 push callback 成功 parse/dispatch。
160    /// 更新 `last_push_received_at_ms` + clear `consecutive_parse_errors`
161    /// + increment `total_pushes_received`.
162    ///
163    /// 若之前 circuit tripped 且 cooldown 过了 → auto-reset（F4 自动恢复路径）。
164    pub fn record_push_success(&self) {
165        let now_ms = unix_now_ms();
166        self.last_push_received_at_ms
167            .store(now_ms, Ordering::Relaxed);
168        self.consecutive_parse_errors.store(0, Ordering::Relaxed);
169        self.total_pushes_received.fetch_add(1, Ordering::Relaxed);
170        // F4 auto-recover: 若 circuit tripped 但已过 cooldown (30s), 自动 reset
171        if self.is_circuit_tripped.load(Ordering::Relaxed) {
172            let tripped_at = self.circuit_breaker_tripped_at_ms.load(Ordering::Relaxed);
173            if tripped_at > 0 && now_ms.saturating_sub(tripped_at) > 30_000 {
174                // 收到成功 push → circuit 显然已恢复
175                self.is_circuit_tripped.store(false, Ordering::Relaxed);
176            }
177        }
178    }
179
180    /// Record a regular QOT push dispatch watermark by FTAPI SubType.
181    ///
182    /// This intentionally lives beside, not inside, `record_push_success()`:
183    /// the backend callback only knows CMD6212, while the dispatcher knows the
184    /// decoded FTAPI `sub_type`. Direct first-push cache replays do not call
185    /// this method because they are synthetic cache events, not backend stream
186    /// freshness evidence.
187    pub fn record_qot_push_sub_type(&self, sub_type: i32) {
188        let now_ms = unix_now_ms();
189        let mut guard = match self.last_push_by_sub_type.lock() {
190            Ok(guard) => guard,
191            Err(poisoned) => poisoned.into_inner(),
192        };
193        guard.insert(sub_type, now_ms);
194    }
195
196    #[cfg(test)]
197    pub(super) fn set_consecutive_parse_errors_for_test(&self, value: u32) {
198        self.consecutive_parse_errors
199            .store(value, Ordering::Relaxed);
200    }
201
202    #[cfg(test)]
203    pub(super) fn set_circuit_breaker_tripped_at_ms_for_test(&self, value: i64) {
204        self.circuit_breaker_tripped_at_ms
205            .store(value, Ordering::Relaxed);
206    }
207
208    #[cfg(test)]
209    pub(super) fn set_last_resubscribe_partial_at_ms_for_test(&self, value: i64) {
210        self.last_resubscribe_partial_at_ms
211            .store(value, Ordering::Relaxed);
212    }
213
214    #[cfg(test)]
215    pub(super) fn set_last_push_received_at_ms_for_test(&self, value: i64) {
216        self.last_push_received_at_ms
217            .store(value, Ordering::Relaxed);
218    }
219
220    #[cfg(test)]
221    pub(super) fn set_trade_requery_drop_state_for_test(&self, total: u64, last_drop_ms: i64) {
222        self.trade_requery_queue_dropped_total
223            .store(total, Ordering::Relaxed);
224        self.last_trade_requery_drop_at_ms
225            .store(last_drop_ms, Ordering::Relaxed);
226    }
227
228    /// 调用时机：push parse 失败（proto decode / 字段缺 / 非法值）。
229    /// 返回**当前** consecutive_errors 值（caller 据此决定是否触发 F3/F4）。
230    pub fn record_parse_error(&self) -> u32 {
231        self.total_parse_errors.fetch_add(1, Ordering::Relaxed);
232        self.consecutive_parse_errors
233            .fetch_add(1, Ordering::Relaxed)
234            + 1
235    }
236
237    /// F3 re-subscribe 触发时调用，更新 counter + 时间戳 + clear 计数.
238    pub fn record_resubscribe(&self) {
239        let now_ms = unix_now_ms();
240        self.resubscribe_triggers.fetch_add(1, Ordering::Relaxed);
241        self.last_resubscribe_at_ms.store(now_ms, Ordering::Relaxed);
242        // re-subscribe 后 push 流应该重建, 清零计数给新流一个 fresh start
243        self.consecutive_parse_errors.store(0, Ordering::Relaxed);
244    }
245
246    /// **v1.4.106 codex 0631 F4 [P2]**: resubscribe **部分失败**触发时调用.
247    /// `missed_keys` = cache miss / unresolved sec_keys 数 (0 = 全部 resolve OK).
248    /// dispatcher / health 检查器 据本字段判 push stream degraded.
249    pub fn record_resubscribe_partial(&self, missed_keys: u32) {
250        let now_ms = unix_now_ms();
251        self.last_resubscribe_partial_at_ms
252            .store(now_ms, Ordering::Relaxed);
253        self.resubscribe_partial_total
254            .fetch_add(1, Ordering::Relaxed);
255        self.last_missed_keys_count
256            .store(missed_keys, Ordering::Relaxed);
257    }
258
259    /// F4 circuit breaker trip 触发时调用.
260    pub fn trip_circuit(&self) {
261        let now_ms = unix_now_ms();
262        self.circuit_breaker_tripped_at_ms
263            .store(now_ms, Ordering::Relaxed);
264        self.circuit_breaker_trips.fetch_add(1, Ordering::Relaxed);
265        self.is_circuit_tripped.store(true, Ordering::Relaxed);
266    }
267
268    /// F4 lightweight state read for threshold checks that must not trigger the
269    /// cooldown auto-reset side effect in `should_skip_dispatch()`.
270    pub fn is_circuit_tripped_now(&self) -> bool {
271        self.is_circuit_tripped.load(Ordering::Relaxed)
272    }
273
274    /// F4 check: dispatcher 每个 event 开头调，tripped 且 cooldown 内 → skip
275    /// dispatch return true (跳过); tripped 但 cooldown 过 → auto-reset return
276    /// false; 未 tripped → return false.
277    pub fn should_skip_dispatch(&self) -> bool {
278        if !self.is_circuit_tripped.load(Ordering::Relaxed) {
279            return false;
280        }
281        let tripped_at = self.circuit_breaker_tripped_at_ms.load(Ordering::Relaxed);
282        if tripped_at == 0 {
283            self.is_circuit_tripped.store(false, Ordering::Relaxed);
284            return false;
285        }
286        let now_ms = unix_now_ms();
287        if now_ms.saturating_sub(tripped_at) > 30_000 {
288            // cooldown 过了, auto-reset
289            self.is_circuit_tripped.store(false, Ordering::Relaxed);
290            false
291        } else {
292            true
293        }
294    }
295
296    /// F6 orphan scan 发现 N 个卡住订单时调（N > 0 才调）.
297    pub fn record_orphan_scan(&self, orphan_count: u64) {
298        let now_ms = unix_now_ms();
299        self.last_orphan_scan_at_ms.store(now_ms, Ordering::Relaxed);
300        if orphan_count > 0 {
301            self.orphan_orders_detected
302                .fetch_add(orphan_count, Ordering::Relaxed);
303        }
304    }
305
306    /// v1.4.106 codex 0932 F1: 调用时机 = `push_tx.try_send(TradeReQuery { .. })`
307    /// 返 `Err(TrySendError::Full | Closed)`. 全部 silent drop 路径都改为
308    /// 调用本方法 + bump counter + log warn 1 次.
309    ///
310    /// `reason` = "queue_full" / "queue_closed";
311    /// `notice_type` = backend NotifyMsg.notice_type (4 / 6 / 100 / 101 / etc.)
312    ///
313    /// 返新的 dropped_total 值 — caller 可据此判定 "首次 drop 触发 warn"
314    /// 但本实装由 caller 简单 log_warn (caller 已有 acc_id / order_ids context).
315    pub fn record_trade_requery_dropped(&self, reason: &'static str, notice_type: u32) -> u64 {
316        let now_ms = unix_now_ms();
317        self.last_trade_requery_drop_at_ms
318            .store(now_ms, Ordering::Relaxed);
319        let n = self
320            .trade_requery_queue_dropped_total
321            .fetch_add(1, Ordering::Relaxed)
322            + 1;
323        // 第一次 drop 立即 warn; 后续 every 100 一次降噪 (避免持续阻塞时刷屏).
324        if n == 1 || n.is_multiple_of(100) {
325            tracing::warn!(
326                reason,
327                notice_type,
328                dropped_total = n,
329                "v1.4.106 audit 0932 F1: TradeReQuery dropped (queue full / closed) — \
330                 backend notify lost; downstream cache may stale until next push or \
331                 manual re-query"
332            );
333        } else {
334            tracing::debug!(
335                reason,
336                notice_type,
337                dropped_total = n,
338                "v1.4.106 audit 0932 F1: TradeReQuery drop (suppressed warn cadence)"
339            );
340        }
341        n
342    }
343
344    /// F5 暴露给 `/api/push-subscriber-info`: 是否 push stream 健康.
345    ///
346    /// 规则：**同时**满足才判 healthy:
347    /// - circuit 未 tripped
348    /// - `consecutive_parse_errors < 5`
349    /// - `last_push_received_at_ms` != 0 且 `now - last < 60_000` (60s 内有 push)
350    ///   **OR** daemon 刚启动 < 60s（没 push 不等于不健康，可能还没订阅）
351    /// - v1.4.106 codex 0932 F1: 60s 内 **没** 发生 TradeReQuery 队列丢弃
352    ///   (队列满 = backpressure, 短期内 dispatcher 无法消费完, 上层应感知)
353    pub fn is_healthy(&self) -> bool {
354        if self.is_circuit_tripped.load(Ordering::Relaxed) {
355            return false;
356        }
357        if self.consecutive_parse_errors.load(Ordering::Relaxed) >= 5 {
358            return false;
359        }
360        // v1.4.106 codex 0932 F1: TradeReQuery drop 60s 内 → degraded
361        let last_drop = self.last_trade_requery_drop_at_ms.load(Ordering::Relaxed);
362        if last_drop != 0 {
363            let now_ms = unix_now_ms();
364            if now_ms.saturating_sub(last_drop) < 60_000 {
365                return false;
366            }
367        }
368        // **v1.4.106 codex 0631 F4 [P2]**: resubscribe partial 60s 内 → degraded.
369        // 表示 reconnect 后部分订阅没复原 (cache miss / backend reject 部分市场).
370        let last_partial = self.last_resubscribe_partial_at_ms.load(Ordering::Relaxed);
371        if last_partial != 0 {
372            let now_ms = unix_now_ms();
373            if now_ms.saturating_sub(last_partial) < 60_000 {
374                return false;
375            }
376        }
377        let last = self.last_push_received_at_ms.load(Ordering::Relaxed);
378        if last == 0 {
379            // 还没 push: 允许 (可能刚启动 / 还没订阅)
380            return true;
381        }
382        let now_ms = unix_now_ms();
383        // 距最近 push < 60s = 健康; >= 60s = 可疑 (可能断流)
384        now_ms.saturating_sub(last) < 60_000
385    }
386
387    /// **v1.4.106 codex 0631 F4 [P2]**: 计算 degraded reason — 与 `is_healthy()`
388    /// 同源逻辑, 但返人类可读字符串 (供 REST `/api/push-subscriber-info`
389    /// 用户排查 / Prometheus alert label).
390    ///
391    /// 优先级 (与 `is_healthy()` 一致):
392    /// 1. circuit_breaker tripped
393    /// 2. consecutive parse errors >= 5
394    /// 3. trade requery queue dropped < 60s
395    /// 4. resubscribe partial failure < 60s
396    /// 5. no push received >= 60s
397    pub fn compute_degraded_reason(&self) -> Option<String> {
398        if self.is_circuit_tripped.load(Ordering::Relaxed) {
399            return Some("circuit_breaker tripped".to_string());
400        }
401        let pe = self.consecutive_parse_errors.load(Ordering::Relaxed);
402        if pe >= 5 {
403            return Some(format!("consecutive parse errors >= 5 (current={pe})"));
404        }
405        let last_drop = self.last_trade_requery_drop_at_ms.load(Ordering::Relaxed);
406        if last_drop != 0 {
407            let now_ms = unix_now_ms();
408            if now_ms.saturating_sub(last_drop) < 60_000 {
409                return Some("trade requery queue dropped within 60s (backpressure)".to_string());
410            }
411        }
412        let last_partial = self.last_resubscribe_partial_at_ms.load(Ordering::Relaxed);
413        if last_partial != 0 {
414            let now_ms = unix_now_ms();
415            if now_ms.saturating_sub(last_partial) < 60_000 {
416                let missed = self.last_missed_keys_count.load(Ordering::Relaxed);
417                return Some(format!(
418                    "resubscribe partial failure within 60s (missed={missed})"
419                ));
420            }
421        }
422        let last_push = self.last_push_received_at_ms.load(Ordering::Relaxed);
423        if last_push == 0 {
424            // 还没 push: 不算 degraded (可能刚启动).
425            return None;
426        }
427        let now_ms = unix_now_ms();
428        if now_ms.saturating_sub(last_push) >= 60_000 {
429            let stale_secs = (now_ms.saturating_sub(last_push)) / 1000;
430            return Some(format!("no push received within 60s (stale={stale_secs}s)"));
431        }
432        None
433    }
434
435    /// F5 snapshot: REST / 诊断层批量读取用.
436    pub fn snapshot(&self) -> PushHealthSnapshot {
437        let mut last_push_by_sub_type: Vec<PushSubTypeWatermark> = {
438            let guard = match self.last_push_by_sub_type.lock() {
439                Ok(guard) => guard,
440                Err(poisoned) => poisoned.into_inner(),
441            };
442            guard
443                .iter()
444                .map(
445                    |(&sub_type, &last_push_received_at_ms)| PushSubTypeWatermark {
446                        sub_type,
447                        last_push_received_at_ms,
448                    },
449                )
450                .collect()
451        };
452        last_push_by_sub_type.sort_by_key(|item| item.sub_type);
453        PushHealthSnapshot {
454            push_stream_healthy: self.is_healthy(),
455            last_push_received_at_ms: self.last_push_received_at_ms.load(Ordering::Relaxed),
456            last_push_by_sub_type,
457            consecutive_parse_errors: self.consecutive_parse_errors.load(Ordering::Relaxed),
458            total_parse_errors: self.total_parse_errors.load(Ordering::Relaxed),
459            total_pushes_received: self.total_pushes_received.load(Ordering::Relaxed),
460            resubscribe_triggers: self.resubscribe_triggers.load(Ordering::Relaxed),
461            circuit_breaker_trips: self.circuit_breaker_trips.load(Ordering::Relaxed),
462            circuit_breaker_tripped_at_ms: self
463                .circuit_breaker_tripped_at_ms
464                .load(Ordering::Relaxed),
465            is_circuit_tripped_now: self.is_circuit_tripped.load(Ordering::Relaxed),
466            orphan_orders_detected: self.orphan_orders_detected.load(Ordering::Relaxed),
467            last_orphan_scan_at_ms: self.last_orphan_scan_at_ms.load(Ordering::Relaxed),
468            last_resubscribe_at_ms: self.last_resubscribe_at_ms.load(Ordering::Relaxed),
469            f2_circuit_tripped_total: self.f2_circuit_tripped_total.load(Ordering::Relaxed),
470            trade_requery_queue_dropped_total: self
471                .trade_requery_queue_dropped_total
472                .load(Ordering::Relaxed),
473            last_trade_requery_drop_at_ms: self
474                .last_trade_requery_drop_at_ms
475                .load(Ordering::Relaxed),
476            last_resubscribe_partial_at_ms: self
477                .last_resubscribe_partial_at_ms
478                .load(Ordering::Relaxed),
479            resubscribe_partial_total: self.resubscribe_partial_total.load(Ordering::Relaxed),
480            last_missed_keys_count: self.last_missed_keys_count.load(Ordering::Relaxed),
481            degraded_reason: self.compute_degraded_reason(),
482        }
483    }
484
485    /// Render push-health state as Prometheus text exposition.
486    ///
487    /// This mirrors the existing `/api/push-subscriber-info` snapshot instead
488    /// of introducing another metrics state source.
489    #[must_use]
490    pub fn render_prometheus(&self) -> String {
491        self.snapshot().render_prometheus()
492    }
493}
494
495/// 共享指针——方便 push callback / dispatcher / REST handler 多处持有.
496pub type SharedPushHealth = Arc<PushHealth>;
497
498pub fn unix_now_ms() -> i64 {
499    crate::bridge::clock::wall_clock_since_unix_epoch_or_zero("push health unix_now_ms").as_millis()
500        as i64
501}
502
503/// v1.4.97 P1-D-B: parse `FUTU_QOT_RELOGIN_BACKOFF_MS` env var.
504///
505/// Format: 4 comma-separated u64 ms values for `[fail_0, fail_1, fail_2,
506/// fail_>=3]` ladder. Default `[60_000, 120_000, 240_000, 600_000]` (60s/2min/
507/// 4min/10min cap, unchanged from v1.4.94).
508///
509/// Parse-fail (wrong count, non-u64, etc.) → emit WARN log + return default.
510/// Empty / unset → return default silently.
511///
512/// Pub for unit testability.
513pub fn parse_relogin_backoff_env(raw: Option<String>) -> [u64; 4] {
514    const DEFAULT: [u64; 4] = [60_000, 120_000, 240_000, 600_000];
515    let Some(s) = raw else { return DEFAULT };
516    let s_trim = s.trim();
517    if s_trim.is_empty() {
518        return DEFAULT;
519    }
520    let parts: Vec<&str> = s_trim.split(',').map(str::trim).collect();
521    if parts.len() != 4 {
522        tracing::warn!(
523            raw = %s_trim,
524            count = parts.len(),
525            "v1.4.97 P1-D-B: FUTU_QOT_RELOGIN_BACKOFF_MS expects 4 comma-separated u64 ms values, using default [60000, 120000, 240000, 600000]"
526        );
527        return DEFAULT;
528    }
529    let mut out = [0u64; 4];
530    for (i, p) in parts.iter().enumerate() {
531        match p.parse::<u64>() {
532            Ok(v) if v > 0 => out[i] = v,
533            _ => {
534                tracing::warn!(
535                    raw = %s_trim,
536                    index = i,
537                    bad_value = *p,
538                    "v1.4.97 P1-D-B: FUTU_QOT_RELOGIN_BACKOFF_MS invalid value at index {i}, using default",
539                );
540                return DEFAULT;
541            }
542        }
543    }
544    out
545}
futu_gateway_core/bridge/push_health/push.rs

futu_gateway_core/bridge/push_health/
push.rs
