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::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::time::SystemTime;

use arc_swap::ArcSwapOption;
use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use parking_lot::Mutex;
use tokio::net::TcpStream;
#[cfg(test)]
use tokio::sync::oneshot;
use tokio::sync::{Mutex as AsyncMutex, mpsc, watch};
use tokio_util::codec::Framed;

use futu_core::error::{FutuError, Result};
use futu_core::log_redact::endpoint_log_fingerprint;

#[cfg(test)]
use crate::nn_codec::NNCodec;
use crate::nn_codec::NNFrame;

mod diagnostics;
mod drop_impl;
mod inbound;
mod lifecycle;
mod outbound;
mod protocol_identity;
mod request;
#[cfg(feature = "test-util")]
mod test_util;

#[cfg(test)]
use diagnostics::InboundFrameStage;
use diagnostics::{
    EndpointFingerprint, InboundProgress, ObservedNNCodec, PendingFailureKind,
    TcpLoginTransportStage, trace_tcp_login_transport_stage,
};
#[cfg(test)]
use diagnostics::{PendingFailureFacts, PendingRegistrationIdentity, PendingResponseEntry};

use inbound::{InboundFrameDecision, decode_inbound_frame_body, is_ex_head_error};
#[cfg(any(test, feature = "test-util"))]
pub use lifecycle::LifecyclePauseHook;
#[cfg(test)]
use lifecycle::PendingRegistrationGuard;
use lifecycle::{
    ConnectionLifecycle, ConnectionTerminationFacts, DiagnosticSink, PendingResponses,
    claim_termination_and_fail_all_pending, emit_diagnostic, mark_disconnected, no_diagnostic_sink,
    pending_failure_facts,
};
pub use protocol_identity::BackendProtocolIdentity;
pub(crate) use request::{BackendWriterAdmission, RequestTimeoutPolicy, WriterAdmissionObserver};

static NEXT_CONNECTION_GENERATION: AtomicU64 = AtomicU64::new(1);

/// 后端连接
pub struct BackendConn {
    serial_no: AtomicU32,
    sec_data: AtomicU32,
    /// Serialize secure-number allocation through writer-queue admission.
    ///
    /// C++ holds `FTChannelImpl::channel_mutex_` across `GetSecureNum()`,
    /// encryption and TCP send/queue admission. Without the same boundary,
    /// concurrent Rust callers can put a lower `sec_data` frame on the wire
    /// after a higher one and trigger backend `-106 security number invalid`.
    /// The guard is released before waiting for the response, so requests
    /// remain concurrently in flight.
    ///
    /// Ref: `f3c/FTNet/Src/ftnet/channel/impl/channel.cpp:907-910,1002-1086`.
    outbound_order: AsyncMutex<()>,
    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: SharedSessionKey,
    /// Previous key retained across an in-place session-key rotation. C++ tries
    /// current then previous when an encrypted response races the refresh.
    previous_session_key: SharedSessionKey,
    cmd_tx: mpsc::Sender<BackendCmd>,
    pending: PendingResponses,
    termination_lifecycle: Arc<ConnectionLifecycle>,
    endpoint_fingerprint: EndpointFingerprint,
    connection_generation: u64,
    inbound_progress: Arc<InboundProgress>,
    diagnostic_sink: DiagnosticSink,
    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>,
    protocol_identity: BackendProtocolIdentity,
    pub client_ver: u16,
}

enum BackendCmd {
    Send {
        frame: NNFrame,
        writer_admitted: Option<Arc<AtomicBool>>,
    },
}
type SharedSessionKey = Arc<ArcSwapOption<Vec<u8>>>;

/// 后端推送回调
/// Backend push callback with the producing connection generation attached.
///
/// The generation is part of the delivery identity: callers must not infer a
/// push source by probing whichever connection happens to be current after the
/// frame was received.
pub type PushCallback = Arc<dyn Fn(u64, u16, Bytes, SystemTime) + Send + Sync + 'static>;

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 固定客户端版本号。
    ///
    /// C++ 同时把 `AppConfig::GetClientVersion()` 写入 protocol header 与
    /// `ReqEncryptData.client_ex_ver`；为便于后台识别，Rust OpenD 于 2026-07-14
    /// 将识别版本提升为 1031。该身份在首次建连和鉴权请求发出前必须已知，
    /// 不能从服务端动态配置取得；尚未识别 1031 的 backend 环境可能拒绝登录。
    /// 若后台重新分配版本，
    /// 只更新本常量，并由契约测试保证所有 wire/HTTP 出口同步变化。
    /// Ref: `f3c/FTNet/Src/ftnet/channel/impl/protocol_header.cpp:25-32` and
    /// `f3c/FTlogin/Src/ftlogin/login/logger.cpp:304-310`.
    pub const CLIENT_VER_FTGTW: u16 = 1031;

    /// 建立底层 TcpStream —— 带超时 + set_nodelay。不 spawn 任何 task。
    async fn establish_stream(addr: &str, timeout: std::time::Duration) -> Result<TcpStream> {
        let addr_fp = endpoint_log_fingerprint(addr);
        let stream = match tokio::time::timeout(timeout, TcpStream::connect(addr)).await {
            Ok(Ok(s)) => s,
            Ok(Err(error)) => {
                return Err(FutuError::Network(std::io::Error::new(
                    error.kind(),
                    format!("backend_connect_failed endpoint_fingerprint={addr_fp}"),
                )));
            }
            Err(_elapsed) => {
                return Err(FutuError::Network(std::io::Error::new(
                    std::io::ErrorKind::TimedOut,
                    format!("backend_connect_timeout endpoint_fingerprint={addr_fp}"),
                )));
            }
        };
        stream.set_nodelay(true).map_err(|error| {
            FutuError::Network(std::io::Error::new(
                error.kind(),
                format!("backend_socket_config_failed endpoint_fingerprint={addr_fp}"),
            ))
        })?;
        Ok(stream)
    }

    /// 连接到后端服务器（带 10s 超时）
    pub async fn connect(
        addr: &str,
        push_callback: PushCallback,
        protocol_identity: BackendProtocolIdentity,
    ) -> 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,
            EndpointFingerprint::from_endpoint(addr),
            protocol_identity,
        ))
    }

    /// 并发连接多个候选地址，谁先通用谁（对齐 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,
        protocol_identity: BackendProtocolIdentity,
    ) -> 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,
                        EndpointFingerprint::from_endpoint(&addr),
                        protocol_identity,
                    );
                    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,
        protocol_identity: BackendProtocolIdentity,
    ) -> Self {
        Self::from_stream_inner(stream, push_callback, protocol_identity)
    }

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

    /// v1.4.70 D1: 泛型化的 stream → BackendConn 构造（内部实现），
    /// 生产路径用 `TcpStream`，test 用 `DuplexStream`。
    #[cfg(any(test, feature = "test-util"))]
    pub(crate) fn from_stream_inner<S>(
        stream: S,
        push_callback: PushCallback,
        protocol_identity: BackendProtocolIdentity,
    ) -> Self
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        Self::from_stream_with_connection_diagnostics(
            stream,
            push_callback,
            EndpointFingerprint::from_endpoint("test-peer"),
            protocol_identity,
            no_diagnostic_sink(),
        )
    }

    #[cfg(test)]
    pub(crate) fn from_stream_inner_default_futunn_chs_for_test<S>(
        stream: S,
        push_callback: PushCallback,
    ) -> Self
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        Self::from_stream_inner(stream, push_callback, BackendProtocolIdentity::new(40, 0))
    }

    /// Construct a production connection around an already-established
    /// transport while immediately sealing its endpoint behind the sole
    /// redaction boundary. The raw endpoint is never stored on `BackendConn`.
    pub(crate) fn from_stream_with_endpoint<S>(
        stream: S,
        push_callback: PushCallback,
        endpoint: &str,
        protocol_identity: BackendProtocolIdentity,
    ) -> Self
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        Self::from_stream_with_connection_diagnostics(
            stream,
            push_callback,
            EndpointFingerprint::from_endpoint(endpoint),
            protocol_identity,
            no_diagnostic_sink(),
        )
    }

    #[cfg(test)]
    pub(crate) fn from_stream_inner_with_diagnostics_default_futunn_chs_for_test<S>(
        stream: S,
        push_callback: PushCallback,
    ) -> (Self, mpsc::Receiver<PendingFailureFacts>)
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        const DIAGNOSTIC_CAPACITY: usize = 16;
        let (diagnostic_tx, diagnostic_rx) = mpsc::channel(DIAGNOSTIC_CAPACITY);
        let conn = Self::from_stream_with_connection_diagnostics(
            stream,
            push_callback,
            EndpointFingerprint::from_endpoint("test-peer"),
            BackendProtocolIdentity::new(40, 0),
            Some(diagnostic_tx),
        );
        (conn, diagnostic_rx)
    }

    fn from_stream_with_connection_diagnostics<S>(
        stream: S,
        push_callback: PushCallback,
        endpoint_fingerprint: EndpointFingerprint,
        protocol_identity: BackendProtocolIdentity,
        diagnostic_sink: DiagnosticSink,
    ) -> Self
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        let connection_generation = NEXT_CONNECTION_GENERATION.fetch_add(1, Ordering::Relaxed);
        let codec = ObservedNNCodec::new_with_connection_identity(
            endpoint_fingerprint.clone(),
            connection_generation,
        );
        let inbound_progress = codec.inbound_progress();
        let framed = Framed::new(stream, codec);
        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::new()));
        let pending_recv = pending.clone();
        let pending_send = pending.clone();
        let termination_lifecycle = Arc::new(ConnectionLifecycle::new());
        let termination_lifecycle_recv = Arc::clone(&termination_lifecycle);
        let termination_lifecycle_send = Arc::clone(&termination_lifecycle);
        let endpoint_fingerprint_recv = endpoint_fingerprint.clone();
        let endpoint_fingerprint_send = endpoint_fingerprint.clone();
        let inbound_progress_recv = Arc::clone(&inbound_progress);
        let inbound_progress_send = Arc::clone(&inbound_progress);
        let diagnostic_sink_recv = diagnostic_sink.clone();
        let diagnostic_sink_send = diagnostic_sink.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 shutdown_tx_recv = shutdown_tx.clone();
        let shutdown_tx_send = shutdown_tx.clone();
        let session_key: SharedSessionKey = Arc::new(ArcSwapOption::empty());
        let session_key_for_recv = session_key.clone();
        let previous_session_key: SharedSessionKey = Arc::new(ArcSwapOption::empty());
        let previous_session_key_for_recv = previous_session_key.clone();

        // 接收任务
        tokio::spawn(async move {
            let termination = loop {
                let next_frame = tokio::select! {
                    _ = shutdown_rx_recv.changed() => {
                        break ConnectionTerminationFacts {
                            kind: PendingFailureKind::Shutdown,
                            endpoint_fingerprint: endpoint_fingerprint_recv.clone(),
                            connection_generation,
                            progress: inbound_progress_recv.snapshot(),
                            diagnostic_sink: diagnostic_sink_recv.clone(),
                        };
                    }
                    next = stream_rx.next() => next,
                };
                let mut frame = match next_frame {
                    Some(Ok(frame)) => frame,
                    Some(Err(_error)) => {
                        break ConnectionTerminationFacts {
                            kind: PendingFailureKind::CodecError,
                            endpoint_fingerprint: endpoint_fingerprint_recv.clone(),
                            connection_generation,
                            progress: inbound_progress_recv.snapshot(),
                            diagnostic_sink: diagnostic_sink_recv.clone(),
                        };
                    }
                    None => {
                        break ConnectionTerminationFacts {
                            kind: PendingFailureKind::PeerEof,
                            endpoint_fingerprint: endpoint_fingerprint_recv.clone(),
                            connection_generation,
                            progress: inbound_progress_recv.snapshot(),
                            diagnostic_sink: diagnostic_sink_recv.clone(),
                        };
                    }
                };

                trace_tcp_login_transport_stage(
                    frame.header.cmd_id,
                    frame.header.serial_no,
                    TcpLoginTransportStage::FrameComplete,
                );

                // 解密（非登录命令）
                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()
                    && is_ex_head_error(&err)
                {
                    // `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"
                        );
                    }
                }

                // C++ NNTCPConnBase first consumes the serial and then dispatches
                // the actual response cmd (NNTCPConnBase.cpp:327-335, 356-364).
                // This Rust pending entry owns one request-specific parser, so a
                // different cmd cannot safely be delivered through that oneshot.
                // Reject the identity mismatch before `serial_matched` instead of
                // mislabelling the request cmd or parsing the body as another API.
                if !frame.header.is_push {
                    let mismatched_entry = {
                        let mut pending = pending_recv.lock();
                        match pending.get(&frame.header.serial_no) {
                            Some(entry) if entry.cmd_id != frame.header.cmd_id => {
                                pending.remove(&frame.header.serial_no)
                            }
                            _ => None,
                        }
                    };
                    if let Some(entry) = mismatched_entry {
                        let mut progress = inbound_progress_recv.snapshot();
                        progress.codec_category = Some("response_cmd_mismatch");
                        let facts = pending_failure_facts(
                            PendingFailureKind::CodecError,
                            entry.cmd_id,
                            entry.serial_no,
                            entry.writer_admitted.load(Ordering::Acquire),
                            &endpoint_fingerprint_recv,
                            connection_generation,
                            &progress,
                        )
                        .with_response_cmd_id(frame.header.cmd_id);
                        emit_diagnostic(&diagnostic_sink_recv, &facts);
                        if entry.tx.send(Err(facts)).is_err() {
                            tracing::debug!(
                                cmd_id = entry.cmd_id,
                                response_cmd_id = frame.header.cmd_id,
                                serial_no = entry.serial_no,
                                "pending response receiver dropped before cmd mismatch delivery"
                            );
                        }
                        continue;
                    }
                }

                let session_key = session_key_for_recv.load_full();
                let previous_session_key = previous_session_key_for_recv.load_full();
                if decode_inbound_frame_body(
                    &mut frame,
                    session_key.as_deref().map(Vec::as_slice),
                    previous_session_key.as_deref().map(Vec::as_slice),
                ) == InboundFrameDecision::Drop
                {
                    if !frame.header.is_push {
                        let mut progress = inbound_progress_recv.snapshot();
                        progress.codec_category = Some("decode_error");
                        if let Some(entry) = pending_recv.lock().remove(&frame.header.serial_no) {
                            trace_tcp_login_transport_stage(
                                entry.cmd_id,
                                entry.serial_no,
                                TcpLoginTransportStage::SerialMatched,
                            );
                            let facts = pending_failure_facts(
                                PendingFailureKind::CodecError,
                                entry.cmd_id,
                                entry.serial_no,
                                entry.writer_admitted.load(Ordering::Acquire),
                                &endpoint_fingerprint_recv,
                                connection_generation,
                                &progress,
                            );
                            emit_diagnostic(&diagnostic_sink_recv, &facts);
                            if entry.tx.send(Err(facts)).is_err() {
                                tracing::debug!(
                                    cmd_id = entry.cmd_id,
                                    serial_no = entry.serial_no,
                                    "pending response receiver dropped before inbound decode failure delivery"
                                );
                            }
                        } else {
                            let facts = pending_failure_facts(
                                PendingFailureKind::UnmatchedSerial,
                                frame.header.cmd_id,
                                frame.header.serial_no,
                                false,
                                &endpoint_fingerprint_recv,
                                connection_generation,
                                &progress,
                            );
                            emit_diagnostic(&diagnostic_sink_recv, &facts);
                        }
                    }
                    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(
                        connection_generation,
                        frame.header.cmd_id,
                        frame.body,
                        SystemTime::now(),
                    );
                } else {
                    let entry = pending_recv.lock().remove(&frame.header.serial_no);
                    if let Some(entry) = entry {
                        trace_tcp_login_transport_stage(
                            entry.cmd_id,
                            entry.serial_no,
                            TcpLoginTransportStage::SerialMatched,
                        );
                        if let Err(Ok(_frame)) = entry.tx.send(Ok(frame)) {
                            tracing::debug!(
                                cmd_id = entry.cmd_id,
                                serial_no = entry.serial_no,
                                "backend response receiver dropped before frame delivery"
                            );
                        }
                    } else {
                        let progress = inbound_progress_recv.snapshot();
                        let facts = pending_failure_facts(
                            PendingFailureKind::UnmatchedSerial,
                            frame.header.cmd_id,
                            frame.header.serial_no,
                            false,
                            &endpoint_fingerprint_recv,
                            connection_generation,
                            &progress,
                        );
                        emit_diagnostic(&diagnostic_sink_recv, &facts);
                    }
                }
            };
            // 连接断开 — 主动回复所有 pending 请求 (C++ OnDisConnectRelpyAll)
            claim_termination_and_fail_all_pending(
                &termination_lifecycle_recv,
                &connected_recv,
                &connected_tx_recv,
                &shutdown_tx_recv,
                &pending_recv,
                &termination,
            );
        });

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

                match cmd {
                    BackendCmd::Send {
                        frame,
                        writer_admitted,
                    } => {
                        // Receiving the command proves queue admission completed. Set the
                        // shared fact before touching the sink so a same-poll send failure
                        // cannot race the request task's post-send store.
                        if let Some(writer_admitted) = writer_admitted {
                            writer_admitted.store(true, Ordering::Release);
                            trace_tcp_login_transport_stage(
                                frame.header.cmd_id,
                                frame.header.serial_no,
                                TcpLoginTransportStage::WriterAdmitted,
                            );
                        }
                        if sink.send(frame).await.is_err() {
                            drop(writer_activity);
                            let termination = ConnectionTerminationFacts {
                                kind: PendingFailureKind::SendError,
                                endpoint_fingerprint: endpoint_fingerprint_send.clone(),
                                connection_generation,
                                progress: inbound_progress_send.snapshot(),
                                diagnostic_sink: diagnostic_sink_send.clone(),
                            };
                            claim_termination_and_fail_all_pending(
                                &termination_lifecycle_send,
                                &connected_send,
                                &connected_tx_send,
                                &shutdown_tx_send,
                                &pending_send,
                                &termination,
                            );
                            break;
                        }
                    }
                }
            }
        });

        Self {
            serial_no: AtomicU32::new(0),
            sec_data: AtomicU32::new(1),
            outbound_order: AsyncMutex::new(()),
            connected,
            connected_tx,
            session_key, // 与接收任务共享同一个 Arc
            previous_session_key,
            cmd_tx,
            pending,
            termination_lifecycle,
            endpoint_fingerprint,
            connection_generation,
            inbound_progress,
            diagnostic_sink,
            shutdown_tx,
            user_id: AtomicU32::new(0),
            client_ip: Mutex::new(String::new()),
            protocol_identity,
            client_ver: Self::CLIENT_VER_FTGTW,
        }
    }

    pub const fn protocol_identity(&self) -> BackendProtocolIdentity {
        self.protocol_identity
    }

    /// 设置 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>) {
        let previous = self.session_key.swap(Some(Arc::new(key)));
        self.previous_session_key.store(previous);
    }

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

    /// 应用登录响应里的 sec_data。
    ///
    /// C++ 只有在 `RspEncryptData.has_sec_data()` 时更新 `secure_num_`；
    /// 字段缺失时保留当前计数，避免 reconnect/login 响应裁剪时把安全数回退。
    ///
    /// Ref: `f3c/FTlogin/Src/ftlogin/login/logger.cpp:553-555`.
    pub fn apply_login_sec_data(&self, val: Option<u32>) {
        if let Some(val) = val {
            self.set_sec_data(val);
        }
    }

    /// 设置登录响应里的客户端外网 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()
    }

    /// Immutable identity of this transport instance, included with every
    /// backend push so consumers can reject delivery from a retired route.
    pub fn connection_generation(&self) -> u64 {
        self.connection_generation
    }

    #[cfg(any(test, feature = "test-util"))]
    pub fn pause_writer_admission_for_test(&self) -> LifecyclePauseHook {
        let hook = LifecyclePauseHook::default();
        *self.termination_lifecycle.admission_publish_hook.lock() = Some(hook.clone());
        hook
    }

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

    pub(crate) fn has_session_key(&self) -> bool {
        self.session_key
            .load_full()
            .is_some_and(|key| !key.is_empty())
    }

    /// 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()
    }

    /// Mark this connection disconnected so owner-side reconnect loops stop
    /// routing business requests through a stale session.
    ///
    /// Used when a higher-level channel event invalidates the session semantics
    /// even if the TCP socket has not closed yet. C++ channel state transitions
    /// publish disconnect/re-establish events; Rust exposes the same liveness
    /// edge through the existing connection-state watch channel.
    pub fn mark_disconnected_for_reconnect(&self, reason: &'static str) {
        tracing::warn!(
            reason,
            "marking backend connection disconnected to trigger reconnect"
        );
        let termination = ConnectionTerminationFacts {
            // This is a local lifecycle invalidation, not peer EOF, codec
            // failure, or writer failure. It shares the approved local
            // shutdown reason with `shutdown_tx` termination.
            kind: PendingFailureKind::Shutdown,
            endpoint_fingerprint: self.endpoint_fingerprint.clone(),
            connection_generation: self.connection_generation,
            progress: self.inbound_progress.snapshot(),
            diagnostic_sink: self.diagnostic_sink.clone(),
        };
        claim_termination_and_fail_all_pending(
            &self.termination_lifecycle,
            &self.connected,
            &self.connected_tx,
            &self.shutdown_tx,
            &self.pending,
            &termination,
        );
    }
}

#[cfg(test)]
mod tests;