futu_gateway_core/bridge/push_health/

qot.rs

//! bridge/push_health/qot — QotLoginHealth + SharedQotLoginHealth
//! (v1.4.110 CC Batch P: 拆自 push_health.rs L655-990)
//!
//! v1.4.90 P1-D / v1.4.118 Jackie: qot-login self-heal state holder.
//!
//! Under extreme network pressure (push storm, partial network failure during
//! heartbeat windows, backend session expiration), the daemon can enter
//! `LoginCache.is_logged_in=false` while TCP is still up. v1.4.83 §9 F2/F3/F4
//! self-heal pipelines cover the push channel (parse errors, dispatcher
//! backpressure, orphan orders), while this module tracks the coarser
//! auth/query-layer signal. Backend disconnect recovery is owned by Phase4
//! reconnect, so this loop does not empty-fire auth refresh while TCP is down.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, AtomicU64, Ordering};

use super::push::{
    RELOGIN_CIRCUIT_COOLDOWN_MS, RELOGIN_CIRCUIT_TRIP_THRESHOLD, parse_relogin_backoff_env,
    unix_now_ms,
};

pub struct QotLoginHealth {
    /// Most recent observation of `qot_logined=false` (Unix ms; 0 = never observed).
    last_qot_logined_false_at_ms: AtomicI64,
    /// Most recent attempt to trigger relogin (Unix ms; 0 = never).
    last_relogin_attempt_at_ms: AtomicI64,
    /// Most recent successful relogin completion (Unix ms; 0 = never).
    last_relogin_success_at_ms: AtomicI64,
    /// Consecutive failed relogin attempts (cleared on success).
    consecutive_relogin_failures: AtomicU32,
    /// Total relogin attempts (monotonic, for metrics rate).
    total_relogin_attempts: AtomicU64,
    /// Total successful relogin completions (monotonic).
    total_relogin_successes: AtomicU64,
    /// Total observations of `qot_logined=false` (monotonic, for metrics).
    total_qot_logined_false_observed: AtomicU64,

    // ========================================================================
    // v1.4.92 P1-D Tier A2: state machine — single in-flight + circuit-open
    // ========================================================================
    /// v1.4.92: 单 in-flight relogin attempt 标志.
    ///
    /// `try_begin_attempt()` 用 `compare_exchange(false, true, AcqRel, Relaxed)`
    /// 在 attempt 之前; `end_attempt()` 用 `store(false, Release)` 之后. 防止 30s
    /// tick 与 attempt 重叠导致两个并发 refresher 同时 POST /authority/.
    ///
    /// 单 in-flight 而不是 N=2/3 并发, 因为 backend 同 uid 短时间内重 POST
    /// /authority/ → 反刷 ret_type=15 (CLAUDE.md auth 坑 #15 第 2 类来源).
    relogin_attempt_in_flight: AtomicBool,

    /// v1.4.92: 5 连续失败后 trip circuit, 跳后续 attempt 10 min.
    ///
    /// 与 F4 push circuit-breaker 思路对齐 (v1.4.83 §9 Phase 2):
    /// - 连续 N=5 次失败 → `open_circuit()` 把 flag 置 true + 记 timestamp
    /// - `should_attempt_relogin_v2` 看到 flag=true + 未超 10 min cooldown → return false
    /// - 10 min 后 `maybe_close_circuit()` 自动重置 flag + 清 failure counter
    ///   (给重试 fresh budget)
    relogin_circuit_open: AtomicBool,

    /// v1.4.92: circuit_open 起始时间 (Unix ms; 0 = 未 open).
    /// 用于判断 cooldown 是否过期.
    relogin_circuit_opened_at_ms: AtomicI64,

    // ========================================================================
    // v1.4.97 P1-D-B: env-tunable backoff ladder
    // ========================================================================
    /// v1.4.97: backoff ladder (ms) per `consecutive_relogin_failures` count
    /// `[fail_0, fail_1, fail_2, fail_>=3]`. Default `[60_000, 120_000,
    /// 240_000, 600_000]` (60s/2min/4min/10min cap, unchanged from v1.4.94).
    ///
    /// Override via env `FUTU_QOT_RELOGIN_BACKOFF_MS=60000,120000,240000,600000`
    /// (4 comma-separated u64 ms). Parse-fail → keep default + WARN log.
    ///
    /// Tester real-machine verify: shorten via env (e.g. 5000,10000,20000,40000)
    /// to cover full ladder in ~1 min instead of waiting 7+ min for default.
    backoff_ms_ladder: [u64; 4],
}

impl Default for QotLoginHealth {
    fn default() -> Self {
        Self::new()
    }
}

impl QotLoginHealth {
    pub fn new() -> Self {
        // v1.4.97 P1-D-B: parse env once at construction time.
        let ladder = parse_relogin_backoff_env(std::env::var("FUTU_QOT_RELOGIN_BACKOFF_MS").ok());
        Self {
            last_qot_logined_false_at_ms: AtomicI64::new(0),
            last_relogin_attempt_at_ms: AtomicI64::new(0),
            last_relogin_success_at_ms: AtomicI64::new(0),
            consecutive_relogin_failures: AtomicU32::new(0),
            total_relogin_attempts: AtomicU64::new(0),
            total_relogin_successes: AtomicU64::new(0),
            total_qot_logined_false_observed: AtomicU64::new(0),
            // v1.4.92 P1-D Tier A2: state machine init
            relogin_attempt_in_flight: AtomicBool::new(false),
            relogin_circuit_open: AtomicBool::new(false),
            relogin_circuit_opened_at_ms: AtomicI64::new(0),
            // v1.4.97 P1-D-B: env-tunable ladder
            backoff_ms_ladder: ladder,
        }
    }

    /// v1.4.97 P1-D-B: test-friendly construction with explicit ladder.
    pub fn with_ladder(ladder: [u64; 4]) -> Self {
        let mut h = Self::new();
        h.backoff_ms_ladder = ladder;
        h
    }

    #[cfg(test)]
    pub(super) fn set_last_relogin_attempt_at_ms_for_test(&self, value: i64) {
        self.last_relogin_attempt_at_ms
            .store(value, Ordering::Relaxed);
    }

    #[cfg(test)]
    pub(super) fn set_relogin_circuit_opened_at_ms_for_test(&self, value: i64) {
        self.relogin_circuit_opened_at_ms
            .store(value, Ordering::Relaxed);
    }

    pub fn backoff_ms_ladder(&self) -> [u64; 4] {
        self.backoff_ms_ladder
    }

    /// Call from query handler / health-poll task whenever
    /// `LoginCache::is_logged_in()` returns `false` while backend is connected.
    /// Updates last-observed timestamp + monotonic counter.
    pub fn record_qot_logined_false(&self) {
        let now_ms = unix_now_ms();
        self.last_qot_logined_false_at_ms
            .store(now_ms, Ordering::Relaxed);
        self.total_qot_logined_false_observed
            .fetch_add(1, Ordering::Relaxed);
    }

    /// Call right before invoking the relogin code path. Records timestamp +
    /// bumps `total_relogin_attempts`.
    pub fn record_relogin_attempt(&self) {
        let now_ms = unix_now_ms();
        self.last_relogin_attempt_at_ms
            .store(now_ms, Ordering::Relaxed);
        self.total_relogin_attempts.fetch_add(1, Ordering::Relaxed);
    }

    /// Call after relogin completes successfully. Clears failure counter,
    /// records timestamp, bumps success monotonic.
    pub fn record_relogin_success(&self) {
        let now_ms = unix_now_ms();
        self.last_relogin_success_at_ms
            .store(now_ms, Ordering::Relaxed);
        self.consecutive_relogin_failures
            .store(0, Ordering::Relaxed);
        self.total_relogin_successes.fetch_add(1, Ordering::Relaxed);
    }

    /// Call after relogin attempt fails. Bumps consecutive failure counter
    /// (drives exponential backoff in `should_attempt_relogin`).
    pub fn record_relogin_failure(&self) {
        self.consecutive_relogin_failures
            .fetch_add(1, Ordering::Relaxed);
    }

    /// **Decision API** for the 30s polling task.
    ///
    /// Returns `true` iff:
    /// 1. caller observed a stale QOT login while the backend TCP is connected:
    ///    `qot_logined_now=false` AND `backend_connected=true`; AND
    /// 2. enough time has passed since `last_relogin_attempt_at_ms` per
    ///    exponential-backoff schedule:
    ///    - 0 failures → 60s
    ///    - 1 failure → 120s
    ///    - 2 failures → 240s
    ///    - 3+ failures → 600s cap
    ///
    /// Caller is responsible for separately calling
    /// `record_relogin_attempt()` then `record_relogin_success()` /
    /// `record_relogin_failure()` after the relogin runs.
    pub fn should_attempt_relogin(&self, qot_logined_now: bool, backend_connected: bool) -> bool {
        if !backend_connected {
            return false;
        }
        if qot_logined_now {
            return false;
        }

        let last_attempt = self.last_relogin_attempt_at_ms.load(Ordering::Relaxed);
        if last_attempt == 0 {
            // Never attempted — fire immediately.
            return true;
        }
        let backoff_ms = self.relogin_backoff_ms();
        let now_ms = unix_now_ms();
        now_ms.saturating_sub(last_attempt) >= backoff_ms
    }

    /// Current backoff window in ms based on `consecutive_relogin_failures`.
    /// Public for tests / metrics; not intended for caller decision (use
    /// `should_attempt_relogin`).
    ///
    /// v1.4.97 P1-D-B: ladder is now env-tunable via
    /// `FUTU_QOT_RELOGIN_BACKOFF_MS=60000,120000,240000,600000`. Default
    /// values unchanged from v1.4.94 baseline.
    pub fn relogin_backoff_ms(&self) -> i64 {
        let failures = self.consecutive_relogin_failures.load(Ordering::Relaxed);
        let idx = (failures.min(3)) as usize;
        self.backoff_ms_ladder[idx] as i64
    }

    // ========================================================================
    // v1.4.92 P1-D Tier A2: state machine API
    // ========================================================================

    /// v1.4.92: 尝试 acquire in-flight slot. 返 `true` 表示获取成功 (caller
    /// 现在可调 refresher), 返 `false` 表示已有别的 attempt 在跑 (caller skip).
    ///
    /// 用 `compare_exchange(false, true, AcqRel, Relaxed)` 而非 `load + store`
    /// 防止两个并发 tick 都看到 false → 都进 attempt.
    pub fn try_begin_attempt(&self) -> bool {
        self.relogin_attempt_in_flight
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
            .is_ok()
    }

    /// v1.4.92: release in-flight slot (无论 attempt 成功或失败都该调).
    pub fn end_attempt(&self) {
        self.relogin_attempt_in_flight
            .store(false, Ordering::Release);
    }

    /// v1.4.92: 触发 circuit_open (5 连续失败后调).
    /// 记 timestamp + 置 flag.
    pub fn open_circuit(&self) {
        let now = unix_now_ms();
        self.relogin_circuit_opened_at_ms
            .store(now, Ordering::Relaxed);
        self.relogin_circuit_open.store(true, Ordering::Release);
    }

    /// v1.4.92: 自动检查 + 重置 circuit (cooldown 过期则关闭 + 清 failure
    /// counter 给重试 fresh budget).
    ///
    /// 每次 polling tick 调一次, 不会引入额外开销 (atomic load + 比较).
    pub fn maybe_close_circuit(&self) {
        if !self.relogin_circuit_open.load(Ordering::Acquire) {
            return;
        }
        let opened_at = self.relogin_circuit_opened_at_ms.load(Ordering::Relaxed);
        if opened_at == 0 {
            // open_circuit 未正确设 timestamp — 直接关闭 (defensive).
            self.relogin_circuit_open.store(false, Ordering::Release);
            return;
        }
        let now = unix_now_ms();
        if now.saturating_sub(opened_at) >= RELOGIN_CIRCUIT_COOLDOWN_MS {
            self.relogin_circuit_open.store(false, Ordering::Release);
            self.relogin_circuit_opened_at_ms
                .store(0, Ordering::Relaxed);
            // 清 failure counter, 让下次 attempt 有 fresh 60s backoff.
            self.consecutive_relogin_failures
                .store(0, Ordering::Relaxed);
            tracing::info!(
                "v1.4.92 P1-D: relogin circuit closed after {}ms cooldown, failure \
                 counter reset to 0",
                RELOGIN_CIRCUIT_COOLDOWN_MS
            );
        }
    }

    /// v1.4.92: 扩展 decision API — 包含 circuit_open 检查.
    ///
    /// 与 [`Self::should_attempt_relogin`] 关系:
    /// - 先 `maybe_close_circuit()` 检查 cooldown 是否过期 → 自动重置
    /// - 若 circuit 仍 open → return false (跳所有 attempt)
    /// - 否则 delegate 到 `should_attempt_relogin` (原 v1.4.90 backoff 逻辑)
    ///
    /// 调用方 (`run_qot_login_health_loop`) 应该用这个 v2 API 而不是 v1
    /// 的 `should_attempt_relogin`. v1 保留作 backward-compat (单测仍在用).
    pub fn should_attempt_relogin_v2(
        &self,
        qot_logined_now: bool,
        backend_connected: bool,
    ) -> bool {
        self.maybe_close_circuit();
        if self.relogin_circuit_open.load(Ordering::Acquire) {
            return false;
        }
        self.should_attempt_relogin(qot_logined_now, backend_connected)
    }

    /// v1.4.92: 扩展版 record_relogin_failure — 失败后检查是否达 trip 阈值.
    ///
    /// 与 [`Self::record_relogin_failure`] 关系:
    /// - 先调 record_relogin_failure (++ counter)
    /// - 若 counter >= [`RELOGIN_CIRCUIT_TRIP_THRESHOLD`] → 自动 open_circuit
    ///
    /// caller 可选用 v1 (只 ++counter) 或 v2 (++ + 触发 trip). polling loop
    /// 实装应该用 v2.
    pub fn record_relogin_failure_with_circuit_check(&self) {
        self.record_relogin_failure();
        let failures = self.consecutive_relogin_failures.load(Ordering::Relaxed);
        if failures >= RELOGIN_CIRCUIT_TRIP_THRESHOLD
            && !self.relogin_circuit_open.load(Ordering::Acquire)
        {
            self.open_circuit();
            tracing::warn!(
                consecutive_failures = failures,
                cooldown_ms = RELOGIN_CIRCUIT_COOLDOWN_MS,
                "v1.4.92 P1-D: relogin circuit OPEN ({} consecutive failures), \
                 skipping all attempts for cooldown window",
                failures
            );
        }
    }

    /// Snapshot for `/api/push-subscriber-info` extension or future
    /// `/api/admin/qot-login-health` endpoint.
    pub fn snapshot(&self) -> QotLoginHealthSnapshot {
        QotLoginHealthSnapshot {
            last_qot_logined_false_at_ms: self.last_qot_logined_false_at_ms.load(Ordering::Relaxed),
            last_relogin_attempt_at_ms: self.last_relogin_attempt_at_ms.load(Ordering::Relaxed),
            last_relogin_success_at_ms: self.last_relogin_success_at_ms.load(Ordering::Relaxed),
            consecutive_relogin_failures: self.consecutive_relogin_failures.load(Ordering::Relaxed),
            total_relogin_attempts: self.total_relogin_attempts.load(Ordering::Relaxed),
            total_relogin_successes: self.total_relogin_successes.load(Ordering::Relaxed),
            total_qot_logined_false_observed: self
                .total_qot_logined_false_observed
                .load(Ordering::Relaxed),
            current_backoff_ms: self.relogin_backoff_ms(),
            // v1.4.92 P1-D Tier A2: state machine snapshot fields
            relogin_attempt_in_flight: self.relogin_attempt_in_flight.load(Ordering::Relaxed),
            relogin_circuit_open: self.relogin_circuit_open.load(Ordering::Relaxed),
            relogin_circuit_opened_at_ms: self.relogin_circuit_opened_at_ms.load(Ordering::Relaxed),
        }
    }
}

/// v1.4.90 P1-D structured snapshot for REST exposure.
///
/// **v1.4.91 P1-D wiring**: REST handler `/api/push-subscriber-info` 已暴露
/// 此 snapshot 在 `qot_login_health` 字段下 (合并入 push_health snapshot).
#[derive(Debug, Clone, serde::Serialize)]
pub struct QotLoginHealthSnapshot {
    pub last_qot_logined_false_at_ms: i64,
    pub last_relogin_attempt_at_ms: i64,
    pub last_relogin_success_at_ms: i64,
    pub consecutive_relogin_failures: u32,
    pub total_relogin_attempts: u64,
    pub total_relogin_successes: u64,
    pub total_qot_logined_false_observed: u64,
    pub current_backoff_ms: i64,
    /// v1.4.92 P1-D Tier A2: 当前是否有 in-flight relogin attempt.
    pub relogin_attempt_in_flight: bool,
    /// v1.4.92 P1-D Tier A2: circuit breaker 是否 open (跳所有 attempt).
    pub relogin_circuit_open: bool,
    /// v1.4.92 P1-D Tier A2: circuit open 起始时间 (Unix ms; 0 = 未 open).
    pub relogin_circuit_opened_at_ms: i64,
}

/// 共享指针 — bridge spawn 30s polling task + REST handler 共享.
///
/// **v1.4.91 P1-D wiring**: bridge `tokio::spawn run_qot_login_health_loop`
/// + REST `/api/push-subscriber-info` provider closure 都已使用此别名.
pub type SharedQotLoginHealth = Arc<QotLoginHealth>;