futu_backend/
heartbeat.rs

1use std::sync::Arc;
2use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
3
4use bytes::Bytes;
5use futu_command_spec::HeartbeatOperation;
6use futu_core::heartbeat::{
7    HeartbeatChannel, HeartbeatFailureAction, HeartbeatMachine, HeartbeatSuccessFacts,
8};
9use futu_core::server_time::{ServerTimeAnchorStore, ServerTimeAnchorUpdate};
10use prost::Message;
11
12use crate::command_runtime::execute_heartbeat;
13use crate::conn::BackendConn;
14use crate::proto_internal::ft_conn_heart_beat::{HeartBeatReq, HeartBeatRsp};
15
16pub use futu_command_spec::{CMD_HEARTBEAT_BROKER, CMD_HEARTBEAT_PLATFORM};
17
18const DEFAULT_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
19
20pub fn start_heartbeat(conn: Arc<BackendConn>, interval: Duration) -> tokio::task::JoinHandle<()> {
21    start_heartbeat_task(
22        conn,
23        interval,
24        HeartbeatOperation::Platform,
25        HeartbeatChannel::Platform,
26        None,
27    )
28}
29
30pub fn start_heartbeat_with_server_clock(
31    conn: Arc<BackendConn>,
32    interval: Duration,
33    server_time_store: Arc<ServerTimeAnchorStore>,
34) -> tokio::task::JoinHandle<()> {
35    start_heartbeat_task(
36        conn,
37        interval,
38        HeartbeatOperation::Platform,
39        HeartbeatChannel::Platform,
40        Some(server_time_store),
41    )
42}
43
44pub fn start_broker_heartbeat(
45    conn: Arc<BackendConn>,
46    interval: Duration,
47) -> tokio::task::JoinHandle<()> {
48    start_heartbeat_task(
49        conn,
50        interval,
51        HeartbeatOperation::Broker,
52        HeartbeatChannel::Broker,
53        None,
54    )
55}
56
57fn start_heartbeat_task(
58    conn: Arc<BackendConn>,
59    interval: Duration,
60    operation: HeartbeatOperation,
61    channel: HeartbeatChannel,
62    server_time_store: Option<Arc<ServerTimeAnchorStore>>,
63) -> tokio::task::JoinHandle<()> {
64    tokio::spawn(async move {
65        let interval = if interval.is_zero() {
66            // Ref: FTlogin/Src/ftlogin/login/logger.cpp:1549-1557.
67            DEFAULT_HEARTBEAT_INTERVAL
68        } else {
69            interval
70        };
71        let mut ticker = tokio::time::interval(interval);
72        ticker.tick().await;
73        let mut machine = HeartbeatMachine::new(channel);
74
75        loop {
76            ticker.tick().await;
77            if !conn.is_connected() {
78                tracing::debug!(
79                    ?operation,
80                    "heartbeat observed disconnected backend channel"
81                );
82                break;
83            }
84
85            let request_plan = machine.request_plan();
86            let request = HeartBeatReq {
87                pre_time_delay: Some(request_plan.pre_time_delay_ms),
88            };
89            let started_at = Instant::now();
90            let result = execute_heartbeat(
91                conn.as_ref(),
92                operation,
93                Bytes::from(request.encode_to_vec()),
94            )
95            .await;
96            let response_captured_at = Instant::now();
97            let rtt_ms = u64::try_from(
98                response_captured_at
99                    .saturating_duration_since(started_at)
100                    .as_millis(),
101            )
102            .unwrap_or(u64::MAX);
103            let local_recv_unix_micros = local_unix_micros();
104
105            let response = match result {
106                Ok(response) => response,
107                Err(error) => {
108                    tracing::warn!(?operation, %error, "backend heartbeat failed");
109                    match machine.record_failure() {
110                        HeartbeatFailureAction::DisconnectAndStop => {
111                            conn.mark_disconnected_for_reconnect("heartbeat failure");
112                            break;
113                        }
114                    }
115                }
116            };
117
118            let response = match HeartBeatRsp::decode(response.body) {
119                Ok(response) => response,
120                Err(error) => {
121                    // C++ logs malformed heartbeat proto and keeps the channel alive.
122                    // Ref: FTlogin/Src/ftlogin/login/logger.cpp:1617-1621.
123                    tracing::warn!(?operation, %error, "failed to decode heartbeat response");
124                    continue;
125                }
126            };
127
128            let success = machine.record_success(HeartbeatSuccessFacts {
129                rtt_ms,
130                server_time_secs: response.server_time,
131                server_time_usec: response.server_time_usec,
132                local_recv_unix_micros,
133            });
134            if let (Some(projection), Some(server_time_store)) =
135                (success.server_time_projection, server_time_store.as_ref())
136            {
137                server_time_store.store_update(ServerTimeAnchorUpdate {
138                    projection,
139                    captured_at: response_captured_at,
140                });
141            }
142            tracing::trace!(
143                ?operation,
144                rtt_ms,
145                result_code = ?response.result_code,
146                next_pre_time_delay_ms = success.next_pre_time_delay_ms,
147                "backend heartbeat ok"
148            );
149        }
150    })
151}
152
153fn local_unix_micros() -> Option<u64> {
154    match SystemTime::now().duration_since(UNIX_EPOCH) {
155        Ok(elapsed) => u64::try_from(elapsed.as_micros()).ok(),
156        Err(error) => {
157            tracing::warn!(%error, "system clock is before UNIX_EPOCH; heartbeat clock not updated");
158            None
159        }
160    }
161}
futu_backend/heartbeat.rs

futu_backend/
heartbeat.rs
