futu_cache/stock_note_state/
owner.rs

1use std::collections::{BTreeMap, HashMap};
2use std::sync::{Arc, Weak};
3
4use futu_domain_qot_stock_note::{NoteOperation, ProjectedNoteBasic};
5use parking_lot::{Mutex, RwLock};
6
7use super::{StockNoteStore, StockNoteStoreError};
8
9#[derive(Clone, Copy, PartialEq, Eq)]
10pub struct StockNoteIdentity {
11    uid: u64,
12    physical_generation: u64,
13}
14
15impl StockNoteIdentity {
16    #[must_use]
17    pub const fn new(uid: u64, physical_generation: u64) -> Self {
18        Self {
19            uid,
20            physical_generation,
21        }
22    }
23}
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum CommitStatus {
27    Applied,
28    IdentityMismatch,
29    IgnoredLateDisk,
30    StaleNeedsRepull,
31    StalePull,
32    StaleMutation,
33}
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct CommitResult {
37    pub status: CommitStatus,
38    pub applied_revision: Option<u64>,
39    pub persistence_error: Option<String>,
40}
41
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct BindResult {
44    /// Failure to save the retired identity is observable but never permits
45    /// that identity to remain active after a login/generation switch.
46    pub save_error: Option<String>,
47    /// Corrupt/unreadable new-user last-good is observable but never prevents
48    /// the clean in-memory identity from becoming current.
49    pub load_error: Option<String>,
50}
51
52impl CommitResult {
53    const fn status(status: CommitStatus) -> Self {
54        Self {
55            status,
56            applied_revision: None,
57            persistence_error: None,
58        }
59    }
60}
61
62#[derive(Clone, Copy)]
63pub struct FullPullToken {
64    identity: StockNoteIdentity,
65    serial: u64,
66    base_revision: u64,
67}
68
69#[derive(Clone, Copy)]
70pub struct MutationToken {
71    identity: StockNoteIdentity,
72    stock_id: u64,
73    generation: u64,
74    base_revision: u64,
75}
76
77impl MutationToken {
78    #[must_use]
79    pub const fn identity(self) -> StockNoteIdentity {
80        self.identity
81    }
82}
83
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct StockNoteSnapshot {
86    pub revision: u64,
87    pub basics: Vec<ProjectedNoteBasic>,
88    pub network_authoritative: bool,
89}
90
91#[derive(Debug, thiserror::Error)]
92pub enum StockNoteStateError {
93    #[error("stock-note identity is not current")]
94    IdentityMismatch,
95    #[error("a full stock-note pull is already in flight")]
96    FullPullInFlight,
97    #[error("stock-note item is missing or has an invalid stock_id")]
98    InvalidNoteItem,
99    #[error("stock-note item stock_id does not match the mutation target")]
100    StockMismatch,
101    #[error(transparent)]
102    Store(#[from] StockNoteStoreError),
103}
104
105#[derive(Default)]
106struct State {
107    identity: Option<StockNoteIdentity>,
108    revision: u64,
109    basics: BTreeMap<u64, ProjectedNoteBasic>,
110    network_authoritative: bool,
111    next_full_serial: u64,
112    full_inflight: Option<u64>,
113    repull_requested: bool,
114    next_mutation_serial: u64,
115    mutation_waiters: HashMap<u64, MutationWaiter>,
116}
117
118struct MutationWaiter {
119    generation: u64,
120    operation: NoteOperation,
121    expected: ProjectedNoteBasic,
122    confirmed_revision: tokio::sync::watch::Sender<Option<u64>>,
123}
124
125pub struct MutationRegistration {
126    token: MutationToken,
127    confirmed_revision: tokio::sync::watch::Receiver<Option<u64>>,
128    owner: Weak<Mutex<State>>,
129}
130
131impl Drop for MutationRegistration {
132    fn drop(&mut self) {
133        let Some(owner) = self.owner.upgrade() else {
134            return;
135        };
136        remove_matching_waiter(&mut owner.lock(), self.token);
137    }
138}
139
140const MUTATION_LOCK_STRIPES: usize = 64;
141
142impl MutationRegistration {
143    #[must_use]
144    pub const fn token(&self) -> MutationToken {
145        self.token
146    }
147
148    #[must_use]
149    pub fn confirmed_by_push(&self) -> bool {
150        self.confirmed_revision.borrow().is_some()
151    }
152
153    pub async fn wait_for_push(&mut self) -> bool {
154        if self.confirmed_by_push() {
155            return true;
156        }
157        self.confirmed_revision.changed().await.is_ok() && self.confirmed_by_push()
158    }
159}
160
161#[derive(Clone)]
162pub struct StockNoteStateOwner {
163    inner: Arc<Mutex<State>>,
164    store: Arc<RwLock<Option<StockNoteStore>>>,
165    mutation_locks: Arc<[Arc<tokio::sync::Mutex<()>>; MUTATION_LOCK_STRIPES]>,
166}
167
168impl StockNoteStateOwner {
169    #[must_use]
170    pub fn identity_matches(&self, identity: StockNoteIdentity) -> bool {
171        self.inner.lock().identity == Some(identity)
172    }
173
174    #[must_use]
175    pub fn memory_only() -> Self {
176        Self {
177            inner: Arc::new(Mutex::new(State::default())),
178            store: Arc::new(RwLock::new(None)),
179            mutation_locks: Arc::new(std::array::from_fn(|_| {
180                Arc::new(tokio::sync::Mutex::new(()))
181            })),
182        }
183    }
184
185    #[must_use]
186    pub fn with_store(store: StockNoteStore) -> Self {
187        Self {
188            inner: Arc::new(Mutex::new(State::default())),
189            store: Arc::new(RwLock::new(Some(store))),
190            mutation_locks: Arc::new(std::array::from_fn(|_| {
191                Arc::new(tokio::sync::Mutex::new(()))
192            })),
193        }
194    }
195
196    pub fn install_store(&self, store: StockNoteStore) {
197        *self.store.write() = Some(store);
198    }
199
200    pub fn bind_identity(
201        &self,
202        identity: StockNoteIdentity,
203    ) -> Result<BindResult, StockNoteStateError> {
204        let retired = {
205            let mut state = self.inner.lock();
206            let store = self.store.read().clone();
207            let retired = state.identity.map(|old| {
208                (
209                    store,
210                    old,
211                    state.revision,
212                    state.basics.values().cloned().collect::<Vec<_>>(),
213                )
214            });
215            *state = State {
216                identity: Some(identity),
217                ..State::default()
218            };
219            retired
220        };
221        let save_error = retired.and_then(|(store, old, revision, basics)| {
222            store.and_then(|store| {
223                store
224                    .save(old.uid, revision, &basics)
225                    .err()
226                    .map(|error| error.to_string())
227            })
228        });
229        let mut load_error = None;
230        let store = self.store.read().clone();
231        if let Some(store) = store.as_ref() {
232            match store.load(identity.uid) {
233                Ok(Some(loaded)) => {
234                    self.apply_loaded_disk(identity, loaded.revision, loaded.basics);
235                }
236                Ok(None) => {}
237                Err(error) => load_error = Some(error.to_string()),
238            }
239        }
240        Ok(BindResult {
241            save_error,
242            load_error,
243        })
244    }
245
246    pub fn apply_loaded_disk(
247        &self,
248        identity: StockNoteIdentity,
249        revision: u64,
250        basics: Vec<ProjectedNoteBasic>,
251    ) -> CommitStatus {
252        let mut state = self.inner.lock();
253        if state.identity != Some(identity) {
254            return CommitStatus::IdentityMismatch;
255        }
256        if state.network_authoritative || state.revision != 0 {
257            return CommitStatus::IgnoredLateDisk;
258        }
259        state.revision = revision;
260        state.basics = admitted_map(basics);
261        CommitStatus::Applied
262    }
263
264    pub fn begin_full_pull(
265        &self,
266        identity: StockNoteIdentity,
267    ) -> Result<FullPullToken, StockNoteStateError> {
268        let mut state = self.inner.lock();
269        if state.identity != Some(identity) {
270            return Err(StockNoteStateError::IdentityMismatch);
271        }
272        if state.full_inflight.is_some() {
273            return Err(StockNoteStateError::FullPullInFlight);
274        }
275        state.next_full_serial = state.next_full_serial.saturating_add(1);
276        let serial = state.next_full_serial;
277        state.full_inflight = Some(serial);
278        Ok(FullPullToken {
279            identity,
280            serial,
281            base_revision: state.revision,
282        })
283    }
284
285    pub fn complete_full_pull(
286        &self,
287        token: FullPullToken,
288        basics: Vec<ProjectedNoteBasic>,
289    ) -> CommitResult {
290        let mut state = self.inner.lock();
291        if state.identity != Some(token.identity) {
292            return CommitResult::status(CommitStatus::IdentityMismatch);
293        }
294        if state.full_inflight != Some(token.serial) {
295            return CommitResult::status(CommitStatus::StalePull);
296        }
297        state.full_inflight = None;
298        if state.revision != token.base_revision {
299            state.repull_requested = true;
300            return CommitResult::status(CommitStatus::StaleNeedsRepull);
301        }
302        state.basics = admitted_map(basics);
303        state.revision = state.revision.saturating_add(1);
304        state.network_authoritative = true;
305        persist_locked(self.store.read().as_ref(), &state, CommitStatus::Applied)
306    }
307
308    pub fn abort_full_pull(&self, token: FullPullToken) {
309        let mut state = self.inner.lock();
310        if state.identity == Some(token.identity) && state.full_inflight == Some(token.serial) {
311            state.full_inflight = None;
312        }
313    }
314
315    pub fn take_repull_requested(&self) -> bool {
316        let mut state = self.inner.lock();
317        std::mem::take(&mut state.repull_requested)
318    }
319
320    pub fn apply_push(
321        &self,
322        identity: StockNoteIdentity,
323        operation: NoteOperation,
324        basic: Option<ProjectedNoteBasic>,
325    ) -> Result<CommitResult, StockNoteStateError> {
326        let mut state = self.inner.lock();
327        if state.identity != Some(identity) {
328            return Ok(CommitResult::status(CommitStatus::IdentityMismatch));
329        }
330        let basic = basic
331            .filter(admissible)
332            .ok_or(StockNoteStateError::InvalidNoteItem)?;
333        let stock_id = basic.stock_id.value;
334        let changed = match operation {
335            // Ref: pinned Desktop StockRemarkService.cpp:179-186. CREATE uses
336            // unordered_map::emplace and therefore does not overwrite an
337            // existing basic; UPDATE assigns.
338            NoteOperation::Create => {
339                if let std::collections::btree_map::Entry::Vacant(entry) =
340                    state.basics.entry(stock_id)
341                {
342                    entry.insert(basic.clone());
343                    true
344                } else {
345                    false
346                }
347            }
348            NoteOperation::Update => {
349                state.basics.insert(stock_id, basic.clone()).as_ref() != Some(&basic)
350            }
351            NoteOperation::Delete => state.basics.remove(&stock_id).is_some(),
352        };
353        if changed {
354            state.revision = state.revision.saturating_add(1);
355        }
356        state.network_authoritative = true;
357        let result = persist_locked(self.store.read().as_ref(), &state, CommitStatus::Applied);
358        if let Some(waiter) = state.mutation_waiters.get(&stock_id)
359            && waiter.operation == operation
360            && changed
361            && push_matches_waiter(operation, &basic, &waiter.expected)
362        {
363            waiter.confirmed_revision.send_replace(Some(state.revision));
364        }
365        Ok(result)
366    }
367
368    #[must_use]
369    pub fn project_ids(&self, stock_ids: &[u64]) -> Vec<ProjectedNoteBasic> {
370        let state = self.inner.lock();
371        stock_ids
372            .iter()
373            .filter_map(|stock_id| state.basics.get(stock_id).cloned())
374            .collect()
375    }
376
377    pub async fn acquire_mutation_lease(
378        &self,
379        identity: StockNoteIdentity,
380        stock_id: u64,
381    ) -> Result<tokio::sync::OwnedMutexGuard<()>, StockNoteStateError> {
382        if stock_id == 0 {
383            return Err(StockNoteStateError::InvalidNoteItem);
384        }
385        if !self.identity_matches(identity) {
386            return Err(StockNoteStateError::IdentityMismatch);
387        }
388        // Fixed stripes keep attacker-controlled stock IDs from growing a
389        // permanent keyed-lock registry. Collisions only serialize unrelated
390        // notes; correctness remains per-stock and identity-fenced.
391        let stripe = (stock_id
392            ^ identity.uid.rotate_left(17)
393            ^ identity.physical_generation.rotate_left(31)) as usize
394            % MUTATION_LOCK_STRIPES;
395        let lock = Arc::clone(&self.mutation_locks[stripe]);
396        let lease = lock.lock_owned().await;
397        if !self.identity_matches(identity) {
398            return Err(StockNoteStateError::IdentityMismatch);
399        }
400        Ok(lease)
401    }
402
403    pub fn begin_mutation(
404        &self,
405        identity: StockNoteIdentity,
406        operation: NoteOperation,
407        expected: ProjectedNoteBasic,
408    ) -> Result<MutationRegistration, StockNoteStateError> {
409        let mut state = self.inner.lock();
410        if state.identity != Some(identity) {
411            return Err(StockNoteStateError::IdentityMismatch);
412        }
413        let stock_id = expected.stock_id.value;
414        if stock_id == 0 {
415            return Err(StockNoteStateError::InvalidNoteItem);
416        }
417        state.next_mutation_serial = state.next_mutation_serial.saturating_add(1);
418        let generation = state.next_mutation_serial;
419        let token = MutationToken {
420            identity,
421            stock_id,
422            generation,
423            base_revision: state.revision,
424        };
425        let (confirmed_revision, receiver) = tokio::sync::watch::channel(None);
426        state.mutation_waiters.insert(
427            stock_id,
428            MutationWaiter {
429                generation: token.generation,
430                operation,
431                expected,
432                confirmed_revision,
433            },
434        );
435        Ok(MutationRegistration {
436            token,
437            confirmed_revision: receiver,
438            owner: Arc::downgrade(&self.inner),
439        })
440    }
441
442    pub fn finish_mutation(&self, token: MutationToken) {
443        let mut state = self.inner.lock();
444        if state.identity != Some(token.identity) {
445            return;
446        }
447        remove_matching_waiter(&mut state, token);
448    }
449
450    pub fn apply_reconcile(
451        &self,
452        token: MutationToken,
453        basic: Option<ProjectedNoteBasic>,
454    ) -> CommitResult {
455        let mut state = self.inner.lock();
456        if state.identity != Some(token.identity) {
457            return CommitResult::status(CommitStatus::IdentityMismatch);
458        }
459        if !state
460            .mutation_waiters
461            .get(&token.stock_id)
462            .is_some_and(|waiter| waiter.generation == token.generation)
463        {
464            return CommitResult::status(CommitStatus::StaleMutation);
465        }
466        // A matching 9430, full refresh, or other authoritative mutation that
467        // landed after registration wins over the later 9432 observation.
468        let matching_push_revision = state
469            .mutation_waiters
470            .get(&token.stock_id)
471            .filter(|waiter| waiter.generation == token.generation)
472            .and_then(|waiter| *waiter.confirmed_revision.borrow());
473        if state.revision != token.base_revision && matching_push_revision != Some(state.revision) {
474            return CommitResult::status(CommitStatus::StaleMutation);
475        }
476        if let Some(basic) = basic {
477            if !admissible(&basic) || basic.stock_id.value != token.stock_id {
478                return CommitResult::status(CommitStatus::StaleMutation);
479            }
480            state.basics.insert(token.stock_id, basic);
481        } else {
482            state.basics.remove(&token.stock_id);
483        }
484        state.revision = state.revision.saturating_add(1);
485        state.network_authoritative = true;
486        persist_locked(self.store.read().as_ref(), &state, CommitStatus::Applied)
487    }
488
489    #[must_use]
490    pub fn snapshot(&self) -> StockNoteSnapshot {
491        let state = self.inner.lock();
492        StockNoteSnapshot {
493            revision: state.revision,
494            basics: state.basics.values().cloned().collect(),
495            network_authoritative: state.network_authoritative,
496        }
497    }
498}
499
500fn remove_matching_waiter(state: &mut State, token: MutationToken) {
501    if state.identity == Some(token.identity)
502        && state
503            .mutation_waiters
504            .get(&token.stock_id)
505            .is_some_and(|waiter| waiter.generation == token.generation)
506    {
507        state.mutation_waiters.remove(&token.stock_id);
508    }
509}
510
511fn push_matches_waiter(
512    operation: NoteOperation,
513    pushed: &ProjectedNoteBasic,
514    expected: &ProjectedNoteBasic,
515) -> bool {
516    if pushed.stock_id != expected.stock_id {
517        return false;
518    }
519    match operation {
520        NoteOperation::Delete => true,
521        NoteOperation::Create | NoteOperation::Update => {
522            pushed.title == expected.title && pushed.is_display == expected.is_display
523        }
524    }
525}
526
527fn admissible(basic: &ProjectedNoteBasic) -> bool {
528    basic.stock_id.present && basic.stock_id.value != 0
529}
530
531fn admitted_map(basics: Vec<ProjectedNoteBasic>) -> BTreeMap<u64, ProjectedNoteBasic> {
532    basics
533        .into_iter()
534        .filter(admissible)
535        .map(|basic| (basic.stock_id.value, basic))
536        .collect()
537}
538
539fn persist_locked(
540    store: Option<&StockNoteStore>,
541    state: &State,
542    status: CommitStatus,
543) -> CommitResult {
544    let persistence_error = store.and_then(|store| {
545        let identity = state.identity?;
546        let basics = state.basics.values().cloned().collect::<Vec<_>>();
547        store
548            .save(identity.uid, state.revision, &basics)
549            .err()
550            .map(|error| error.to_string())
551    });
552    CommitResult {
553        status,
554        applied_revision: Some(state.revision),
555        persistence_error,
556    }
557}
558
559#[cfg(test)]
560#[path = "owner_tests.rs"]
561mod tests;
futu_cache/stock_note_state/owner.rs

futu_cache/stock_note_state/
owner.rs
