futu_gateway_core/bridge/push_health/push/
f2.rs

1use std::sync::atomic::Ordering;
2
3use super::{PushHealth, unix_now_ms};
4
5// ===== v1.4.90 P2-B: F2 retry exhaust 账户级 circuit-break =====
6//
7// **背景**：双 tester v1.4.89 验证报告 P2-B: 一账户 (FutuMY) CMD3020 稳定失败.
8// F2 单次 4 次 (immediate / 1s / 3s / 9s) 退避全失败 "exhausted" → 30s 后
9// 外层 dispatch 又起完整 F2 (4 次) → 一天 ~10000 条 WARN 刷屏.
10// F3/F4 (parse 错误) 有 circuit-break, F2 (TradeReQuery) 没.
11//
12// **修法**：account-level circuit-break:
13// - 30 min 窗口内同 acc_id F2 exhausted >= 3 次 → trip circuit
14// - tripped acc 在 1h 内 retry skip (caller 检查 is_f2_circuit_tripped 立即 return)
15// - 1h 后 reset (cooldown 过 → 重新允许 retry)
16// - `f2_circuit_tripped_total` AtomicU64 counter (Prometheus 暴露)
17// - log warn 1 次 (trip 瞬间) 而非每秒刷屏
18
19/// F2 失败计数窗口 (30 min)
20pub(in crate::bridge::push_health) const F2_EXHAUST_WINDOW_MS: i64 = 30 * 60 * 1000;
21/// F2 trip 阈值 (窗口内 >= N 次 exhausted)
22pub(in crate::bridge::push_health) const F2_TRIP_THRESHOLD: usize = 3;
23/// F2 trip 后 cooldown (1 hour)
24pub(in crate::bridge::push_health) const F2_COOLDOWN_MS: i64 = 60 * 60 * 1000;
25
26/// 单账户 F2 状态: 窗口内 exhaust 时间戳列表 + tripped_at.
27#[derive(Debug, Default, Clone)]
28pub(super) struct F2AccState {
29    /// 窗口内 exhausted 时间戳 (Unix ms), 按时间序 push, 越界自动 trim.
30    exhausted_at_ms: Vec<i64>,
31    /// 当前 trip 起始时间 (Unix ms, 0 = 未 tripped).
32    tripped_at_ms: i64,
33}
34
35impl F2AccState {
36    /// trim 30 min 之外的 exhausted 时间戳, 保留窗口内.
37    fn trim_window(&mut self, now_ms: i64) {
38        let cutoff = now_ms - F2_EXHAUST_WINDOW_MS;
39        self.exhausted_at_ms.retain(|&t| t >= cutoff);
40    }
41}
42
43impl PushHealth {
44    #[cfg(test)]
45    pub(in crate::bridge::push_health) fn set_f2_tripped_at_ms_for_test(
46        &self,
47        acc_id: u32,
48        tripped_at_ms: i64,
49    ) {
50        if let Ok(mut guard) = self.f2_acc_state.lock() {
51            guard.entry(acc_id).or_default().tripped_at_ms = tripped_at_ms;
52        }
53    }
54
55    #[cfg(test)]
56    pub(in crate::bridge::push_health) fn push_f2_exhausted_at_ms_for_test(
57        &self,
58        acc_id: u32,
59        exhausted_at_ms: i64,
60    ) {
61        if let Ok(mut guard) = self.f2_acc_state.lock() {
62            guard
63                .entry(acc_id)
64                .or_default()
65                .exhausted_at_ms
66                .push(exhausted_at_ms);
67        }
68    }
69
70    #[cfg(test)]
71    pub(in crate::bridge::push_health) fn f2_state_snapshot_for_test(
72        &self,
73        acc_id: u32,
74    ) -> Option<(usize, i64)> {
75        self.f2_acc_state.lock().ok().and_then(|guard| {
76            guard
77                .get(&acc_id)
78                .map(|state| (state.exhausted_at_ms.len(), state.tripped_at_ms))
79        })
80    }
81
82    /// v1.4.90 P2-B: 调用时机 = F2 retry 4 次全失败 (`retry_with_exp_backoff`
83    /// 返 `Err`). 记录该 acc_id 的 exhausted 事件; 若 30 min 窗口内 >= 3 次 →
84    /// trip circuit + return `true` (caller log warn 一次 + ++counter).
85    ///
86    /// 返 false = 未 trip (caller 只 log debug); 返 true = 刚 trip (caller log
87    /// warn + ++ `f2_circuit_tripped_total`).
88    pub fn record_f2_exhausted(&self, acc_id: u32) -> bool {
89        let now_ms = unix_now_ms();
90        let mut guard = match self.f2_acc_state.lock() {
91            Ok(g) => g,
92            Err(poisoned) => poisoned.into_inner(),
93        };
94        let st = guard.entry(acc_id).or_default();
95        st.trim_window(now_ms);
96        st.exhausted_at_ms.push(now_ms);
97        let already_tripped =
98            st.tripped_at_ms != 0 && now_ms.saturating_sub(st.tripped_at_ms) < F2_COOLDOWN_MS;
99        if already_tripped {
100            // 已经 tripped 仍在 cooldown 内, 不重复 bump counter
101            return false;
102        }
103        if st.exhausted_at_ms.len() >= F2_TRIP_THRESHOLD {
104            st.tripped_at_ms = now_ms;
105            // 释放 guard 后再 bump 计数, 避免锁里调 atomic 顺序混乱
106            drop(guard);
107            self.f2_circuit_tripped_total
108                .fetch_add(1, Ordering::Relaxed);
109            return true;
110        }
111        false
112    }
113
114    /// v1.4.90 P2-B: 调用时机 = dispatcher 准备给 acc_id 起 F2 retry 之前.
115    /// 返 true → caller 应 skip 本次 retry (账户在 cooldown 内);
116    /// 返 false → 正常 retry.
117    ///
118    /// 1h cooldown 过则自动 reset (清 tripped_at_ms + 清 exhaust window).
119    pub fn is_f2_circuit_tripped(&self, acc_id: u32) -> bool {
120        let now_ms = unix_now_ms();
121        let mut guard = match self.f2_acc_state.lock() {
122            Ok(g) => g,
123            Err(poisoned) => poisoned.into_inner(),
124        };
125        let Some(st) = guard.get_mut(&acc_id) else {
126            return false;
127        };
128        if st.tripped_at_ms == 0 {
129            return false;
130        }
131        let elapsed = now_ms.saturating_sub(st.tripped_at_ms);
132        if elapsed >= F2_COOLDOWN_MS {
133            // cooldown 过, auto-reset
134            st.tripped_at_ms = 0;
135            st.exhausted_at_ms.clear();
136            return false;
137        }
138        true
139    }
140
141    /// v1.4.90 P2-B: 仅测试用: 强制 reset 某 acc_id 的 F2 circuit.
142    #[cfg(test)]
143    pub fn reset_f2_circuit(&self, acc_id: u32) {
144        if let Ok(mut guard) = self.f2_acc_state.lock()
145            && let Some(st) = guard.get_mut(&acc_id)
146        {
147            st.tripped_at_ms = 0;
148            st.exhausted_at_ms.clear();
149        }
150    }
151}
futu_gateway_core/bridge/push_health/push/f2.rs

futu_gateway_core/bridge/push_health/push/
f2.rs
