futu_cache/trd_cache/
freshness.rs

1use std::hash::Hash;
2use std::sync::atomic::{AtomicU64, Ordering};
3use std::time::Instant;
4
5use dashmap::DashMap;
6use dashmap::mapref::entry::Entry;
7use futu_domain_trade_account::TradeSnapshotFreshnessFacts;
8
9use super::{CachedFunds, CachedPosition, PositionsCacheKey, TrdCache};
10
11#[derive(Debug, Clone)]
12pub struct StampedTradeSnapshot<T> {
13    value: T,
14    updated_at: Instant,
15    sequence: u64,
16}
17
18impl<T> StampedTradeSnapshot<T> {
19    #[must_use]
20    pub fn value(&self) -> &T {
21        &self.value
22    }
23
24    #[must_use]
25    pub fn sequence(&self) -> u64 {
26        self.sequence
27    }
28}
29
30#[derive(Debug, Clone)]
31pub struct FundsSnapshotLookup {
32    pub funds: Option<CachedFunds>,
33    pub currency_match: bool,
34    pub freshness: TradeSnapshotFreshnessFacts,
35}
36
37#[derive(Debug, Clone)]
38pub struct PositionsSnapshotLookup {
39    pub positions: Option<Vec<CachedPosition>>,
40    pub freshness: TradeSnapshotFreshnessFacts,
41    /// Immutable cache publication sequence for request/response fencing.
42    pub sequence: Option<u64>,
43}
44
45pub(super) struct TradeSnapshotFreshnessStore {
46    next_sequence: AtomicU64,
47    asset_push_sequences: DashMap<PositionsCacheKey, u64>,
48}
49
50impl TradeSnapshotFreshnessStore {
51    pub(super) fn new() -> Self {
52        Self {
53            next_sequence: AtomicU64::new(1),
54            asset_push_sequences: DashMap::new(),
55        }
56    }
57
58    pub(super) fn stamp<T>(&self, value: T) -> StampedTradeSnapshot<T> {
59        StampedTradeSnapshot {
60            value,
61            updated_at: Instant::now(),
62            sequence: self.take_sequence(),
63        }
64    }
65
66    pub(super) fn record_asset_push(&self, key: PositionsCacheKey) {
67        let sequence = self.take_sequence();
68        match self.asset_push_sequences.entry(key) {
69            Entry::Occupied(mut entry) => {
70                if sequence > *entry.get() {
71                    entry.insert(sequence);
72                }
73            }
74            Entry::Vacant(entry) => {
75                entry.insert(sequence);
76            }
77        }
78    }
79
80    pub(super) fn freshness<T>(
81        &self,
82        key: PositionsCacheKey,
83        snapshot: Option<&StampedTradeSnapshot<T>>,
84    ) -> TradeSnapshotFreshnessFacts {
85        let Some(snapshot) = snapshot else {
86            return TradeSnapshotFreshnessFacts {
87                snapshot_age_ms: None,
88                predates_server_push: false,
89            };
90        };
91        let snapshot_age_ms =
92            u64::try_from(snapshot.updated_at.elapsed().as_millis()).unwrap_or(u64::MAX);
93        let predates_server_push = self
94            .asset_push_sequences
95            .get(&key)
96            .is_some_and(|push_sequence| snapshot.sequence <= *push_sequence);
97        TradeSnapshotFreshnessFacts {
98            snapshot_age_ms: Some(snapshot_age_ms),
99            predates_server_push,
100        }
101    }
102
103    fn take_sequence(&self) -> u64 {
104        // A process would need more than 2^64 cache/push writes to wrap. Keeping
105        // one total order avoids wall-clock equality and backwards-clock bugs.
106        self.next_sequence.fetch_add(1, Ordering::Relaxed)
107    }
108}
109
110pub(super) fn insert_newer_snapshot<K, T>(
111    map: &DashMap<K, StampedTradeSnapshot<T>>,
112    key: K,
113    snapshot: StampedTradeSnapshot<T>,
114) where
115    K: Eq + Hash,
116{
117    match map.entry(key) {
118        Entry::Occupied(mut entry) => {
119            if snapshot.sequence > entry.get().sequence {
120                entry.insert(snapshot);
121            }
122        }
123        Entry::Vacant(entry) => {
124            entry.insert(snapshot);
125        }
126    }
127}
128
129impl TrdCache {
130    /// Record the C++ `UpdateSvrPushTime(NN_AssetKey)` equivalent.
131    ///
132    /// Ref: `NNData_Trd_Acc.cpp:416-420`. Funds, positions and combo positions
133    /// under the same `(acc_id, asset_category)` compare against this watermark.
134    pub fn mark_asset_server_push(&self, acc_id: u64, asset_category: i32) {
135        self.snapshot_freshness
136            .record_asset_push(PositionsCacheKey::scoped(acc_id, asset_category));
137    }
138}
futu_cache/trd_cache/freshness.rs

futu_cache/trd_cache/
freshness.rs
