futu_gateway_core/

idempotency.rs

//! v1.4.38 Bug-防赔钱：订单幂等 cache
//!
//! # Why
//!
//! 任何在线 trade 服务都需要幂等保护：网络抖动 / SDK auto-retry / LLM agent 幻觉
//! / 脚本 try-except 循环 都可能导致**重复下单** —— 金融后果是真金白银。
//!
//! Futu backend 的 `remark` 字段只 echo-back 不做 dedup，服务端不提供原生幂等。
//! 所以必须在 client（daemon）侧做。
//!
//! # Design
//!
//! **短期去重窗口**：客户端传 `Idempotency-Key` → daemon 查 cache：
//!   - cache hit（TTL 内）→ 返缓存的结果，不触发真下单
//!   - cache miss → 真下单 → 写 cache → 返结果
//!
//! **TTL = 90 秒**：略长于 SDK auto-retry 典型时窗（5-60s）+ LLM agent 幻觉
//! 重 call 窗口（10s-几分钟）。不用 5 分钟是因为交易语义里"同样的 body 5 分钟
//! 后"大概率是**新意图**（scalper 策略信号重触发）不是 retry。
//!
//! **daemon 不生成 key，只读**：没 key 直接透传（backward-compat）。key 由客户端
//! / agent 按业务意图 per-intent 分配; daemon 只识别显式 key, 不从订单参数猜测意图。
//!
//! **cache both success + failure**：幂等语义要求"同 key 重入得同一结果"。如果
//! 用户想重试失败（比如补了保证金）→ 用**新 key**表示新意图，不是重用旧 key。

use std::sync::Arc;
use std::time::{Duration, Instant};

use dashmap::DashMap;
use dashmap::mapref::entry::Entry;

/// 默认 TTL：90 秒
pub const DEFAULT_TTL_SECS: u64 = 90;

/// 交易写路径 PacketID 防重放 guard.
///
/// C++ `IsReplayAttack(nConnID, packetID)` 在 PlaceOrder / ModifyOrder 进入
/// backend 前检查:
/// - `packetID.connID == actual conn_id`
/// - 同连接 `packetID.serialNo` 单调递增
///
/// Rust 另有 idempotency cache 用于正常 retry（同 key 返同一 response）。因此
/// handler 调用顺序应是: 先 idempotency cache hit，再执行本 guard；这样"同
/// PacketID 的 retry"命中 cache，不会被当作 replay 误拒，而"新请求使用旧
/// serial"仍会在 backend 前 fail-closed。
#[derive(Debug, Default)]
pub struct TradeReplayGuard {
    last_serial_by_conn: DashMap<u64, u32>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TradeReplayError {
    ConnMismatch {
        actual_conn_id: u64,
        packet_conn_id: u64,
    },
    SerialReplay {
        conn_id: u64,
        last_serial_no: u32,
        packet_serial_no: u32,
    },
}

impl std::fmt::Display for TradeReplayError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ConnMismatch {
                actual_conn_id,
                packet_conn_id,
            } => write!(
                f,
                "packetID.connID ({packet_conn_id}) 与实际连接 conn_id ({actual_conn_id}) 不一致"
            ),
            Self::SerialReplay {
                conn_id,
                last_serial_no,
                packet_serial_no,
            } => write!(
                f,
                "packetID.serialNo ({packet_serial_no}) 不大于连接 {conn_id} 上次写请求 serialNo ({last_serial_no})"
            ),
        }
    }
}

impl TradeReplayGuard {
    pub fn new() -> Self {
        Self::default()
    }

    /// 检查 FTAPI trade-write `Common.PacketID`.
    ///
    /// `(0, 0)` 视为非 FTAPI binary surface 的兼容路径（例如 REST JSON 调用未
    /// 传 packetID），不参与 replay guard。
    ///
    /// `serialNo == 0` 视为显式 idempotency-key 映射路径：其重放语义是有意
    /// dedup，由幂等 cache 负责，不参与 C++ serial monotonic check。
    pub fn check_packet(
        &self,
        actual_conn_id: u64,
        packet_conn_id: u64,
        packet_serial_no: u32,
    ) -> Result<(), TradeReplayError> {
        if packet_conn_id == 0 && packet_serial_no == 0 {
            return Ok(());
        }
        if packet_serial_no == 0 {
            return Ok(());
        }
        if packet_conn_id != actual_conn_id {
            return Err(TradeReplayError::ConnMismatch {
                actual_conn_id,
                packet_conn_id,
            });
        }

        match self.last_serial_by_conn.entry(actual_conn_id) {
            Entry::Occupied(mut entry) => {
                let last = *entry.get();
                if packet_serial_no <= last {
                    Err(TradeReplayError::SerialReplay {
                        conn_id: actual_conn_id,
                        last_serial_no: last,
                        packet_serial_no,
                    })
                } else {
                    *entry.get_mut() = packet_serial_no;
                    Ok(())
                }
            }
            Entry::Vacant(entry) => {
                if packet_serial_no == 0 {
                    Err(TradeReplayError::SerialReplay {
                        conn_id: actual_conn_id,
                        last_serial_no: 0,
                        packet_serial_no,
                    })
                } else {
                    entry.insert(packet_serial_no);
                    Ok(())
                }
            }
        }
    }
}

/// 已缓存结果的值 — 包含**序列化后的 response body** + 插入时间。
/// 序列化后存：下次 hit 直接返回同一字节流，无需再走 proto encode。
#[derive(Clone)]
struct CacheEntry {
    /// 序列化后的 response body（proto encoded 或 JSON）
    response: Vec<u8>,
    /// 插入时间，用于 TTL 过期
    inserted_at: Instant,
}

/// 订单幂等 cache。线程安全（DashMap + Instant）。
///
/// **v1.4.106 codex 0920 F1 (P1) namespace fix**: cache key 必须 namespace 隔离
/// 跨 (endpoint, caller_key_id, acc_id, op). 否则:
///
/// 1. **Cross-pollination**: PlaceOrder 后 ModifyOrder 用同一 `Idempotency-Key`
///    会拿回 PlaceOrder response (严重交易 bug —— 用户以为改完了但实际没发).
/// 2. **Cross-account leak**: 不同 caller (key_id A vs B) 用同一 key 会跨账户
///    串 response (隐私 + 资金安全).
/// 3. **Cross-acc/env leak**: 同 caller 在 acc_id=A 下单后立刻在 acc_id=B 用
///    同 key 改单, 会拿到 acc_id=A 的 PlaceOrder response.
///
/// **Namespace 设计**: `endpoint|caller|acc_id|op|raw_key`
/// - endpoint = `place_order` / `modify_order` / `cancel_order` / `reconfirm_order`
/// - caller = caller_key_id (Some) 或 `<no_key>` (TCP/legacy)
/// - acc_id = c2s.header.acc_id (u64 decimal)
/// - op = `PLACE` / `MODIFY=N` / `CANCEL` / `RECONFIRM=N` (etc.)
/// - raw_key = 用户 Idempotency-Key 或 `tcp-pkt-{conn_id}-{serial_no}`
///
/// 同时**所有 5 轴**都不同的请求, key 不会撞.
pub struct IdempotencyCache {
    map: DashMap<String, CacheEntry>,
    in_flight: DashMap<String, ()>,
    ttl: Duration,
}

/// v1.4.106 codex 0920 F1 (P1): namespaced cache key builder.
///
/// 所有 trade-write handler 都用本 helper 构造 cache key, 不直接拼字符串.
/// 这样保证 namespace 5 轴的 ordering / separator 全 daemon 统一.
#[derive(Debug, Clone)]
pub struct CacheKey {
    /// endpoint: `place_order` / `modify_order` / `cancel_order` / `reconfirm_order`
    pub endpoint: &'static str,
    /// caller key id (None → `<no_key>` 占位符 for TCP/legacy)
    pub caller_key_id: Option<String>,
    /// acc_id (c2s.header.acc_id)
    pub acc_id: u64,
    /// op label —— `PLACE` / `MODIFY=N` / `CANCEL` / `RECONFIRM=N` 等
    pub op: String,
    /// 用户传的 raw key (Idempotency-Key header / TCP packet id)
    pub raw_key: String,
}

impl CacheKey {
    /// 构造 namespaced 字符串 key, 用 `|` 分隔, caller None → `<no_key>` 占位.
    pub fn build(&self) -> String {
        let caller = self.caller_key_id.as_deref().unwrap_or("<no_key>");
        format!(
            "{ep}|caller={caller}|acc={acc}|op={op}|raw={raw}",
            ep = self.endpoint,
            caller = caller,
            acc = self.acc_id,
            op = self.op,
            raw = self.raw_key,
        )
    }
}

impl IdempotencyCache {
    pub fn new(ttl_secs: u64) -> Self {
        Self {
            map: DashMap::new(),
            in_flight: DashMap::new(),
            ttl: Duration::from_secs(ttl_secs),
        }
    }

    /// 默认 TTL。
    pub fn with_default_ttl() -> Self {
        Self::new(DEFAULT_TTL_SECS)
    }

    /// 查询 cache：
    /// - Some(cached_response)：TTL 内命中，返回缓存结果
    /// - None：未命中 / 已过期（调用方应真执行 + [`Self::put`]）
    ///
    /// ⚠️ 过期的 entry **不在此清理**（避免抢锁），由周期性 [`Self::purge_expired`]
    /// 清。cache hit 判断只看 `inserted_at.elapsed() < ttl`。
    pub fn get(&self, key: &str) -> Option<Vec<u8>> {
        let entry = self.map.get(key)?;
        if entry.inserted_at.elapsed() < self.ttl {
            Some(entry.response.clone())
        } else {
            None
        }
    }

    /// 插入新结果。相同 key 覆盖（但实际上调用方应在 get=None 之后才 put，
    /// 避免并发两次 put 覆盖）。
    pub fn put(&self, key: String, response: Vec<u8>) {
        self.map.insert(
            key,
            CacheEntry {
                response,
                inserted_at: Instant::now(),
            },
        );
    }

    /// Store the terminal response for a key that was guarded by [`Self::begin`].
    ///
    /// Call sites keep the fallback `put` path because older code may still
    /// construct a cache key without acquiring a guard. New trade-write paths
    /// should normally pass the guard returned by `begin`.
    pub fn put_finished_response(
        &self,
        key: &str,
        guard: &mut Option<IdempotencyFlightGuard<'_>>,
        response: Vec<u8>,
    ) {
        if let Some(guard) = guard.take() {
            guard.put_response(response);
        } else {
            self.put(key.to_string(), response);
        }
    }

    /// Per-key singleflight gate for idempotent trade writes.
    ///
    /// The old pattern was `get -> await backend -> put`, so two concurrent
    /// requests with the same key could both miss and both execute the backend
    /// write. This method serializes executions for the same namespaced key:
    /// followers wait until the leader stores a response or drops the guard.
    pub async fn begin(&self, key: String) -> IdempotencyBegin<'_> {
        loop {
            if let Some(cached) = self.get(&key) {
                return IdempotencyBegin::Cached(cached);
            }
            match self.in_flight.entry(key.clone()) {
                dashmap::mapref::entry::Entry::Vacant(v) => {
                    let inserted = v.insert(());
                    drop(inserted);
                    return IdempotencyBegin::Execute(IdempotencyFlightGuard { cache: self, key });
                }
                dashmap::mapref::entry::Entry::Occupied(o) => {
                    drop(o);
                }
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    }

    /// 清理已过期 entry。建议 spawn 一个 tokio::time::interval 每 60s 调一次。
    /// 即使不调也无正确性问题（get 自己会判 TTL），只是 cache 内存会累积。
    pub fn purge_expired(&self) -> usize {
        let ttl = self.ttl;
        let before = self.map.len();
        self.map.retain(|_, v| v.inserted_at.elapsed() < ttl);
        before.saturating_sub(self.map.len())
    }

    /// 当前 cache 大小（含未过期 + 已过期未清理的）。仅用于 metrics。
    pub fn len(&self) -> usize {
        self.map.len()
    }

    pub fn is_empty(&self) -> bool {
        self.map.is_empty()
    }
}

pub enum IdempotencyBegin<'a> {
    Cached(Vec<u8>),
    Execute(IdempotencyFlightGuard<'a>),
}

pub struct IdempotencyFlightGuard<'a> {
    cache: &'a IdempotencyCache,
    key: String,
}

impl IdempotencyFlightGuard<'_> {
    pub fn put_response(self, response: Vec<u8>) {
        self.cache.put(self.key.clone(), response);
    }
}

impl Drop for IdempotencyFlightGuard<'_> {
    fn drop(&mut self) {
        self.cache.in_flight.remove(&self.key);
    }
}

#[cfg(test)]
mod singleflight_tests;

impl Default for IdempotencyCache {
    fn default() -> Self {
        Self::with_default_ttl()
    }
}

/// Spawn 一个后台 task 周期性清理过期 entries。返回的 task handle 交由
/// bridge-owned [`crate::bridge::BackgroundRuntime`] 托管。
pub fn spawn_cleanup_task(cache: Arc<IdempotencyCache>) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(Duration::from_secs(60));
        ticker.tick().await; // 跳过首次立即触发
        loop {
            ticker.tick().await;
            let purged = cache.purge_expired();
            if purged > 0 {
                tracing::debug!(purged, "idempotency cache: purged expired entries");
            }
        }
    })
}

/// v1.4.38 audit log 扩展字段：本次 place-order 请求的 cache 结果。
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub enum CacheOutcome {
    /// 客户端没传 key，没走幂等 path（backward-compat）
    NoKey,
    /// 传了 key，cache hit，返缓存结果（未触发真下单）
    Cached,
    /// 传了 key，cache miss，真执行 + 写 cache
    Executed,
}

impl CacheOutcome {
    pub fn as_str(&self) -> &'static str {
        match self {
            CacheOutcome::NoKey => "no_key",
            CacheOutcome::Cached => "cached",
            CacheOutcome::Executed => "executed",
        }
    }
}

#[cfg(test)]
mod tests;