futu_backend/

conn.rs

// 后端 TCP 连接管理（直连富途后端服务器）
//
// 加密方式：
// - 登录命令(1001/6001/26001): 不加密
// - 其他命令: AES-128 加密，body = encrypt(sec_data(4B BE) + proto_data)

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::SystemTime;

use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use parking_lot::Mutex;
use tokio::net::TcpStream;
use tokio::sync::{mpsc, oneshot, watch};
use tokio_util::codec::Framed;

use futu_core::error::{FutuError, Result};
use futu_core::log_redact::endpoint_log_fingerprint;
use futu_net::encrypt::{aes_cbc_md5_decrypt_var, aes_cbc_md5_encrypt_var};

use crate::nn_codec::{NNCodec, NNFrame, NNHeader, should_skip_encryption};

/// 后端连接
pub struct BackendConn {
    serial_no: AtomicU32,
    sec_data: AtomicU32,
    connected: Arc<std::sync::atomic::AtomicBool>,
    connected_tx: watch::Sender<bool>,
    /// 共享的 session key — 接收任务和发送方共用同一个 Arc。
    /// 对齐 C++ `Logger::session_key_` 是 `std::string`（`logger.h:152`），
    /// 长度由服务端下发决定，Platform 通常 16 字节（AES-128），Broker 可能是
    /// 32 字节（AES-256）。v1.4.7 之前固定 `[u8; 16]` 对 broker session_key
    /// 截断会导致服务端 `CONN decrypt failed`。
    session_key: Arc<Mutex<Option<Vec<u8>>>>,
    cmd_tx: mpsc::Sender<BackendCmd>,
    pending: PendingResponses,
    shutdown_tx: watch::Sender<bool>,
    pub user_id: AtomicU32,
    /// RspEncryptData.client_ip(field 14), set after TCP login.
    ///
    /// C++ `logger.cpp:511-516` stores this server-observed public IP on the
    /// working TCP client, then `logger.cpp:1122-1125` echoes it in CMD20147.
    client_ip: Mutex<String>,
    pub client_type: u8,
    pub client_ver: u16,
    pub lang_id: u8,
}

enum BackendCmd {
    Send(NNFrame),
}

type PendingResponses = Arc<Mutex<HashMap<u32, oneshot::Sender<NNFrame>>>>;

/// 后端推送回调
pub type PushCallback = Arc<dyn Fn(u16, Bytes, SystemTime) + Send + Sync + 'static>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum InboundFrameDecision {
    Deliver,
    Drop,
}

fn decode_inbound_frame_body(
    frame: &mut NNFrame,
    session_key: Option<&[u8]>,
) -> InboundFrameDecision {
    if !should_skip_encryption(frame.header.cmd_id)
        && let Some(key) = session_key
    {
        let body_len = frame.body.len();
        // C++ `logger.cpp:1840-1855` 的约束：加密包 body_len 必须 >= 32
        // 且是 16 字节对齐（AES-CBC-MD5 最小输出 32 字节）。
        // C++ 还有特例：body_len == 32 认为是空包，直接清空。
        if body_len >= 32 && body_len.is_multiple_of(16) {
            match aes_cbc_md5_decrypt_var(key, &frame.body) {
                Ok(decrypted) => {
                    // 后端响应不含 sec_data 前缀（仅 client→server 方向有）
                    frame.body = Bytes::from(decrypted);
                }
                Err(e) => {
                    // C++ `NNTCPConnBase.cpp:347-351` 会先用 current session
                    // key 解密，再用 old session key 重试；两次失败后
                    // `If_OMWarn_ReturnVoid`，不会把原始密文交给业务层。
                    tracing::warn!(
                        cmd_id = frame.header.cmd_id,
                        body_len = body_len,
                        key_len = key.len(),
                        error = %e,
                        "decrypt failed, dropping inbound frame"
                    );
                    return InboundFrameDecision::Drop;
                }
            }
        } else {
            tracing::debug!(
                cmd_id = frame.header.cmd_id,
                body_len = body_len,
                "body not encrypted (len not aligned to 16)"
            );
        }
    }

    if frame.header.is_compressed() {
        let compressed_body_len = frame.body.len();
        match crate::ftlogin_wire::decode_inbound_body(true, frame.body.as_ref()) {
            Ok(decompressed) => {
                frame.body = Bytes::from(decompressed);
                frame.header.body_len = frame.body.len() as u32;
                tracing::debug!(
                    cmd_id = frame.header.cmd_id,
                    serial_no = frame.header.serial_no,
                    compressed_body_len,
                    body_len = frame.body.len(),
                    "decompressed inbound frame after decrypt"
                );
            }
            Err(e) => {
                // C++ FTLogin `channel_impl.cpp:1905-1954` marks
                // decompression errors as `kSdkMsgRecvDataError`; it does not
                // deliver the compressed raw body as a successful business
                // payload.
                tracing::warn!(
                    cmd_id = frame.header.cmd_id,
                    serial_no = frame.header.serial_no,
                    body_len = compressed_body_len,
                    error = %e,
                    "decompress failed after decrypt, dropping inbound frame"
                );
                return InboundFrameDecision::Drop;
            }
        }
    }

    InboundFrameDecision::Deliver
}

fn release_pending_on_inbound_decode_error(frame: &NNFrame, pending: &PendingResponses) {
    if !frame.header.is_push {
        pending.lock().remove(&frame.header.serial_no);
    }
}

fn mark_disconnected(
    connected: &Arc<std::sync::atomic::AtomicBool>,
    connected_tx: &watch::Sender<bool>,
) {
    let was_connected = connected.swap(false, Ordering::AcqRel);
    if was_connected {
        let _ = connected_tx.send(false);
    }
}

impl BackendConn {
    /// 连接 TCP 的超时时间。
    ///
    /// Linux 默认 `tcp_syn_retries=6` 时 `TcpStream::connect` 等到 `ETIMEDOUT`
    /// 需要约 127 秒——用户启动 opend 时如果选到一个不通的 IP 就卡 2 分钟后
    /// 才报错 offline mode（某位 Rocky Linux 用户踩过）。加 10s 超时快速失败，
    /// 让上层（`bridge.rs` 的 connect 循环）有机会 fallback 到下一个候选 IP。
    pub const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

    /// Backend 侧可识别的 Rust OpenD 客户端版本号。
    ///
    /// Rust OpenD 的后台诊断版本号。
    ///
    /// 该值与普通 C++ OpenD 拉开，便于后台排查时区分 Rust 链路。服务端若要求
    /// version >= 800，低版本会返回 `error_code=45 "当前应用版本过低"`。
    pub const CLIENT_VER_FTGTW: u16 = 1030;

    /// 建立底层 TcpStream —— 带超时 + set_nodelay。不 spawn 任何 task。
    async fn establish_stream(addr: &str, timeout: std::time::Duration) -> Result<TcpStream> {
        let stream = match tokio::time::timeout(timeout, TcpStream::connect(addr)).await {
            Ok(Ok(s)) => s,
            Ok(Err(e)) => return Err(e.into()),
            Err(_elapsed) => {
                return Err(FutuError::Network(std::io::Error::new(
                    std::io::ErrorKind::TimedOut,
                    format!("connect to {addr} timed out after {}s", timeout.as_secs()),
                )));
            }
        };
        stream.set_nodelay(true)?;
        Ok(stream)
    }

    /// 连接到后端服务器（带 10s 超时）
    pub async fn connect(addr: &str, push_callback: PushCallback) -> Result<Self> {
        let stream = Self::establish_stream(addr, Self::CONNECT_TIMEOUT).await?;
        tracing::info!(
            addr_fp = %endpoint_log_fingerprint(addr),
            "connected to backend"
        );
        Ok(Self::from_stream(stream, push_callback))
    }

    /// 并发连接多个候选地址，谁先通用谁（对齐 C++ `connector.cpp:175-189`
    /// `ConnectStrategyAddr` 的 concurrency_ip 语义）。
    ///
    /// - 每个候选独立带 `CONNECT_TIMEOUT`（10s）超时
    /// - 第一个 `Ok(stream)` 胜出，其余 pending task drop 时会关闭半连接
    /// - 全部失败返回最后一个错误
    ///
    /// 返回 `(BackendConn, winner_addr)`，调用方用 winner_addr 做登录协议里的
    /// host_ip/host_port 字段。
    pub async fn connect_race(
        addrs: &[String],
        push_callback: PushCallback,
    ) -> Result<(Self, String)> {
        use futures::stream::{FuturesUnordered, StreamExt};

        if addrs.is_empty() {
            return Err(FutuError::Network(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "connect_race: empty address list",
            )));
        }

        tracing::info!(
            candidates = addrs.len(),
            candidate_fps = ?addrs
                .iter()
                .map(|addr| endpoint_log_fingerprint(addr))
                .collect::<Vec<_>>(),
            "racing parallel connects"
        );

        let mut attempts: FuturesUnordered<_> = addrs
            .iter()
            .cloned()
            .map(|addr| async move {
                let result = Self::establish_stream(&addr, Self::CONNECT_TIMEOUT).await;
                (addr, result)
            })
            .collect();

        let mut last_err: Option<FutuError> = None;
        while let Some((addr, result)) = attempts.next().await {
            match result {
                Ok(stream) => {
                    tracing::info!(
                        addr_fp = %endpoint_log_fingerprint(&addr),
                        remaining_losers = attempts.len(),
                        "connect race winner"
                    );
                    drop(attempts); // 其他 FuturesUnordered 里的 future drop 即取消
                    let conn = Self::from_stream(stream, push_callback);
                    return Ok((conn, addr));
                }
                Err(e) => {
                    tracing::debug!(
                        addr_fp = %endpoint_log_fingerprint(&addr),
                        error = %e,
                        "candidate failed"
                    );
                    last_err = Some(e);
                }
            }
        }

        Err(last_err.unwrap_or_else(|| {
            FutuError::Network(std::io::Error::other("connect_race: all candidates failed"))
        }))
    }

    /// v1.4.70 D1: test-only 从 `tokio::io::DuplexStream` 构造 BackendConn
    ///
    /// 用于 integration tests（`crates/futu-gateway/tests/common/mock_backend.rs`）
    /// 替代真 `TcpStream`。生产路径通过 `connect()` → `from_stream()` 不变。
    #[cfg(feature = "test-util")]
    pub fn from_duplex(stream: tokio::io::DuplexStream, push_callback: PushCallback) -> Self {
        Self::from_stream_inner(stream, push_callback)
    }

    /// 从已建立的 TcpStream 构造 BackendConn（spawn recv/send task）
    fn from_stream(stream: TcpStream, push_callback: PushCallback) -> Self {
        Self::from_stream_inner(stream, push_callback)
    }

    /// v1.4.70 D1: 泛型化的 stream → BackendConn 构造（内部实现），
    /// 生产路径用 `TcpStream`，test 用 `DuplexStream`。
    pub(crate) fn from_stream_inner<S>(stream: S, push_callback: PushCallback) -> Self
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        let framed = Framed::new(stream, NNCodec);
        let (mut sink, mut stream_rx) = framed.split();

        let (cmd_tx, mut cmd_rx) = mpsc::channel::<BackendCmd>(256);

        let pending: PendingResponses =
            Arc::new(Mutex::new(HashMap::<u32, oneshot::Sender<NNFrame>>::new()));
        let pending_recv = pending.clone();
        let pending_send = pending.clone();
        let connected = Arc::new(std::sync::atomic::AtomicBool::new(true));
        let (connected_tx, _) = watch::channel(true);
        let (shutdown_tx, mut shutdown_rx_recv) = watch::channel(false);
        let mut shutdown_rx_send = shutdown_tx.subscribe();
        let connected_recv = Arc::clone(&connected);
        let connected_send = Arc::clone(&connected);
        let connected_tx_recv = connected_tx.clone();
        let connected_tx_send = connected_tx.clone();
        let session_key: Arc<Mutex<Option<Vec<u8>>>> = Arc::new(Mutex::new(None));
        let session_key_for_recv = session_key.clone();

        // 接收任务
        tokio::spawn(async move {
            loop {
                let next_frame = tokio::select! {
                    _ = shutdown_rx_recv.changed() => {
                        break;
                    }
                    next = stream_rx.next() => next,
                };
                let Some(Ok(mut frame)) = next_frame else {
                    break;
                };

                // 解密（非登录命令）
                tracing::debug!(
                    cmd_id = frame.header.cmd_id,
                    serial_no = frame.header.serial_no,
                    is_push = frame.header.is_push,
                    is_compressed = frame.header.is_compressed,
                    ex_head_len = frame.header.ex_head_len,
                    body_len = frame.header.body_len,
                    actual_body_len = frame.body.len(),
                    "recv frame"
                );

                // 如果 ex_head 里有业务错误（err_info.cmd_result != 0），
                // 把它 log 出来 —— 服务端 body 空时会通过 ex_head 返回错误。
                //
                // 软失败特判：某些返回 code 是正常的"无此服务/无数据"信号（例如
                // 某些账号在 Futu HK broker 上没有授权账户时 CMD 2298 会收到
                // `code=-102 CONN can not find command service`）。这类日志降到
                // DEBUG 避免噪声；其他错误保持 WARN。
                if let Some(err) = frame.parse_ex_head_error()
                    && (err.cmd_result != 0
                        || err.code != 0
                        || !err.message.is_empty()
                        || !err.source.is_empty())
                {
                    // `code=-102` 是服务端的软失败信号："此账户/通道不支持该 cmd"。
                    // 服务端实际把 "CONN can not find command service" 放在 `source`
                    // 字段里，`message` 为空（和 C++ ErrorInfo 字段 2/4 定义略有偏差）。
                    // 例如：账户在 Futu HK broker 通道不认 CMD 2298 / CMD 1003 heartbeat
                    // 都会走到这里。降级到 debug 避免日志噪声；其他 code 保持 warn。
                    let is_soft_fail = err.code == -102;
                    if is_soft_fail {
                        tracing::debug!(
                            cmd_id = frame.header.cmd_id,
                            code = err.code,
                            source = %err.source,
                            message = %err.message,
                            "server: cmd not available on this channel (soft-fail)"
                        );
                    } else {
                        tracing::warn!(
                            cmd_id = frame.header.cmd_id,
                            cmd_result = err.cmd_result,
                            code = err.code,
                            source = %err.source,
                            message = %err.message,
                            "server returned err_info in ex_head"
                        );
                    }
                }

                let session_key = session_key_for_recv.lock().clone();
                if decode_inbound_frame_body(&mut frame, session_key.as_deref())
                    == InboundFrameDecision::Drop
                {
                    release_pending_on_inbound_decode_error(&frame, &pending_recv);
                    continue;
                }

                // 判断是否为推送帧:
                // 1. flags.push_ == 1 (标准推送)
                // 2. flags.push_ == 0 但 serial_no == 0 (后端首次订阅快照,
                //    以 Reply 形式发送, 如 CMD6212 的初始摆盘/报价)
                let is_push = frame.header.is_push
                    || (frame.header.serial_no == 0 && !pending_recv.lock().contains_key(&0));
                if is_push {
                    tracing::debug!(
                        cmd_id = frame.header.cmd_id,
                        body_len = frame.body.len(),
                        is_push = frame.header.is_push,
                        is_compressed = frame.header.is_compressed,
                        reserved = ?frame.header.reserved,
                        "backend push received"
                    );
                    push_callback(frame.header.cmd_id, frame.body, SystemTime::now());
                } else {
                    let tx = pending_recv.lock().remove(&frame.header.serial_no);
                    if let Some(tx) = tx
                        && let Err(frame) = tx.send(frame)
                    {
                        tracing::debug!(
                            cmd_id = frame.header.cmd_id,
                            serial_no = frame.header.serial_no,
                            body_len = frame.body.len(),
                            "backend response receiver dropped before frame delivery"
                        );
                    }
                }
            }
            // 连接断开 — 清理所有 pending 请求 (C++ OnDisConnectRelpyAll)
            mark_disconnected(&connected_recv, &connected_tx_recv);
            tracing::warn!("backend connection closed");
            let mut pending = pending_recv.lock();
            let count = pending.len();
            if count > 0 {
                tracing::warn!(
                    pending_count = count,
                    "aborting pending requests due to disconnect"
                );
            }
            // 清空 pending HashMap 会 drop 所有 oneshot::Sender
            // 这会导致对应的 oneshot::Receiver 返回 RecvError
            // 从而让 request() 方法返回 "response channel closed" 错误
            pending.clear();
        });

        // 发送任务
        tokio::spawn(async move {
            loop {
                let next_cmd = tokio::select! {
                    _ = shutdown_rx_send.changed() => {
                        break;
                    }
                    cmd = cmd_rx.recv() => cmd,
                };
                let Some(cmd) = next_cmd else {
                    break;
                };

                match cmd {
                    BackendCmd::Send(frame) => {
                        if let Err(e) = sink.send(frame).await {
                            mark_disconnected(&connected_send, &connected_tx_send);
                            tracing::error!(error = %e, "backend send failed");
                            let mut pending = pending_send.lock();
                            let count = pending.len();
                            if count > 0 {
                                tracing::warn!(
                                    pending_count = count,
                                    "aborting pending requests due to send failure"
                                );
                            }
                            pending.clear();
                            break;
                        }
                    }
                }
            }
        });

        Self {
            serial_no: AtomicU32::new(0),
            sec_data: AtomicU32::new(1),
            connected,
            connected_tx,
            session_key, // 与接收任务共享同一个 Arc
            cmd_tx,
            pending,
            shutdown_tx,
            user_id: AtomicU32::new(0),
            client_ip: Mutex::new(String::new()),
            client_type: 40, // NN_ClientType_ApiGateway
            client_ver: Self::CLIENT_VER_FTGTW,
            lang_id: 0,
        }
    }

    /// 设置 session key（登录成功后调用）。接受变长字节，16/24/32 分别对应
    /// AES-128/192/256。对齐 C++ `Logger::session_key_` 是 `std::string`，
    /// 长度取决于服务端下发的 `RspEncryptData.session_key` 字段原始长度。
    pub fn set_session_key(&self, key: Vec<u8>) {
        *self.session_key.lock() = Some(key);
    }

    /// 设置 sec_data 初始值（登录成功后调用）
    pub fn set_sec_data(&self, val: u32) {
        self.sec_data.store(val, Ordering::Relaxed);
    }

    /// 设置登录响应里的客户端外网 IP。
    pub fn set_client_ip(&self, ip: String) {
        *self.client_ip.lock() = ip;
    }

    /// 读取登录响应里的客户端外网 IP。
    pub fn client_ip(&self) -> String {
        self.client_ip.lock().clone()
    }

    pub fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Acquire)
    }

    /// Subscribe to low-level connection liveness changes.
    ///
    /// C++ FTLogin publishes a connection-closed event immediately
    /// (`GTWCmdAndPushReply::OMEvProc_ConnClosed`) and lets the gateway enter
    /// reconnecting state without waiting for a later business request. Rust
    /// uses this watch channel to wake the reconnect monitor as soon as the
    /// recv/send task marks the TCP channel disconnected.
    pub fn subscribe_connection_state(&self) -> watch::Receiver<bool> {
        self.connected_tx.subscribe()
    }

    /// 发送请求并等待响应
    pub async fn request(&self, cmd_id: u16, body: Vec<u8>) -> Result<NNFrame> {
        self.request_with_reserved(cmd_id, body, [0u8; 10]).await
    }

    /// 发送请求并等待响应（带 header reserved 字段，用于行情命令传递市场类型）
    pub async fn request_with_reserved(
        &self,
        cmd_id: u16,
        body: Vec<u8>,
        reserved: [u8; 10],
    ) -> Result<NNFrame> {
        self.request_with_reserved_timeout(
            cmd_id,
            body,
            reserved,
            std::time::Duration::from_secs(10),
        )
        .await
    }

    pub(crate) async fn request_with_reserved_timeout(
        &self,
        cmd_id: u16,
        body: Vec<u8>,
        reserved: [u8; 10],
        timeout: std::time::Duration,
    ) -> Result<NNFrame> {
        let deadline = tokio::time::Instant::now() + timeout;
        let frame = self.build_outbound_frame(cmd_id, body, reserved)?;
        let serial_no = frame.header.serial_no;

        let (resp_tx, resp_rx) = oneshot::channel();
        self.pending.lock().insert(serial_no, resp_tx);
        match tokio::time::timeout_at(deadline, self.cmd_tx.send(BackendCmd::Send(frame))).await {
            Ok(Ok(())) => {}
            Ok(Err(_closed)) => {
                self.pending.lock().remove(&serial_no);
                mark_disconnected(&self.connected, &self.connected_tx);
                return Err(FutuError::NotInitialized);
            }
            Err(_elapsed) => {
                self.pending.lock().remove(&serial_no);
                return Err(FutuError::Timeout);
            }
        }

        let resp = crate::delay_stats::trace_backend_request(cmd_id, async {
            match tokio::time::timeout_at(deadline, resp_rx).await {
                Ok(Ok(resp)) => Ok(resp),
                Ok(Err(_closed)) => {
                    self.pending.lock().remove(&serial_no);
                    Err(FutuError::Codec("response channel closed".into()))
                }
                Err(_elapsed) => {
                    self.pending.lock().remove(&serial_no);
                    Err(FutuError::Timeout)
                }
            }
        })
        .await?;

        Ok(resp)
    }

    /// 发送无需等待响应的消息
    pub async fn send_fire_and_forget(&self, cmd_id: u16, body: Vec<u8>) -> Result<()> {
        let frame = self.build_outbound_frame(cmd_id, body, [0u8; 10])?;
        if self.cmd_tx.send(BackendCmd::Send(frame)).await.is_err() {
            mark_disconnected(&self.connected, &self.connected_tx);
            return Err(FutuError::NotInitialized);
        }

        Ok(())
    }

    fn build_outbound_frame(
        &self,
        cmd_id: u16,
        body: Vec<u8>,
        reserved: [u8; 10],
    ) -> Result<NNFrame> {
        let serial = self.next_serial();
        let mut header = NNHeader::new(cmd_id, serial);
        header.user_id = self.user_id.load(Ordering::Relaxed);
        header.client_type = self.client_type;
        header.client_ver = self.client_ver;
        header.lang_id = self.lang_id;
        // 行情命令实际用到的只有 reserved[0..2]（market_type + ex_type），
        // 后 2 字节（原 [8..10]）在 C++ protocol header 里是 ex_head_len 位置，
        // 我们不发 ex_head 所以这 2 字节固定为 0。
        header.reserved.copy_from_slice(&reserved[..8]);

        let final_body = self.encode_outbound_body(cmd_id, body)?;
        header.body_len = final_body.len() as u32;

        Ok(NNFrame {
            header,
            body: Bytes::from(final_body),
            ex_head: Bytes::new(),
        })
    }

    fn encode_outbound_body(&self, cmd_id: u16, body: Vec<u8>) -> Result<Vec<u8>> {
        if should_skip_encryption(cmd_id) {
            return Ok(body);
        }

        let key = self.session_key.lock().clone();
        match key {
            Some(key) => {
                // C++ 先 m_nSecData++ 再使用，所以要用 +1 后的值。
                let sec = self.sec_data.fetch_add(1, Ordering::Relaxed) + 1;
                let mut plaintext = Vec::with_capacity(4 + body.len());
                plaintext.extend_from_slice(&sec.to_be_bytes());
                plaintext.extend_from_slice(&body);
                // var 版支持 16/24/32 字节 key —— broker session_key 可能是 32 字节。
                aes_cbc_md5_encrypt_var(&key, &plaintext)
            }
            None => Ok(body),
        }
    }

    fn next_serial(&self) -> u32 {
        self.serial_no.fetch_add(1, Ordering::Relaxed) + 1
    }
}

impl Drop for BackendConn {
    fn drop(&mut self) {
        let _ = self.shutdown_tx.send(true);
        mark_disconnected(&self.connected, &self.connected_tx);
        self.pending.lock().clear();
    }
}

#[cfg(test)]
mod tests;