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

futu_cache/trd_cache/
freshness.rs
