Skip to main content

cranpose_core/
snapshot_state_observer.rs

1// Complex types are inherent to the observer pattern with nested callbacks and state tracking
2#![allow(clippy::type_complexity)]
3
4use crate::collections::map::{HashMap, HashSet};
5use crate::hash::default as default_hash;
6use crate::snapshot_v2::{register_apply_observer, ReadObserver, StateObjectId};
7use crate::state::StateObject;
8use crate::{RecomposeScope, RecomposeScopeInner, ScopeId};
9use smallvec::SmallVec;
10use std::any::{Any, TypeId};
11use std::cell::{Cell, RefCell};
12use std::hash::{Hash, Hasher};
13use std::rc::{Rc, Weak};
14use std::sync::Arc;
15
16/// Executes a callback once changes are delivered.
17type Executor = dyn Fn(Box<dyn FnOnce() + 'static>) + 'static;
18
19/// Observer that records state object reads performed inside a given scope and
20/// notifies the caller when any of the observed objects change.
21///
22/// This is a pragmatic Rust translation of Jetpack Compose's
23/// `SnapshotStateObserver`. The implementation focuses on the core behaviour
24/// needed by the Cranpose runtime:
25/// - Tracking state object reads per logical scope.
26/// - Reacting to snapshot apply notifications.
27/// - Scheduling invalidation callbacks via the supplied executor.
28///
29/// Advanced features from the Kotlin version (derived state tracking, change
30/// coalescing, queue minimisation) are deferred
31#[derive(Clone)]
32pub struct SnapshotStateObserver {
33    inner: Rc<SnapshotStateObserverInner>,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
37pub struct SnapshotStateObserverDebugStats {
38    pub scopes_len: usize,
39    pub scopes_cap: usize,
40    pub fast_scopes_len: usize,
41    pub fast_scopes_cap: usize,
42    pub stateless_scope_count: usize,
43    pub observed_state_count: usize,
44    pub observed_state_capacity: usize,
45}
46
47impl SnapshotStateObserver {
48    /// Create a new observer that schedules callbacks using `on_changed_executor`.
49    pub fn new(on_changed_executor: impl Fn(Box<dyn FnOnce() + 'static>) + 'static) -> Self {
50        let inner = Rc::new(SnapshotStateObserverInner::new(on_changed_executor));
51        inner.set_self(Rc::downgrade(&inner));
52        Self { inner }
53    }
54
55    /// Observe state object reads performed while executing `block`.
56    ///
57    /// Subsequent calls to `observe_reads` replace any previously recorded
58    /// observations for the provided `scope`. When one of the observed objects
59    /// mutates, `on_value_changed_for_scope` will be invoked on the executor.
60    pub fn observe_reads<T, R>(
61        &self,
62        scope: T,
63        on_value_changed_for_scope: impl Fn(&T) + 'static,
64        block: impl FnOnce() -> R,
65    ) -> R
66    where
67        T: Any + Clone + Eq + Hash + 'static,
68    {
69        self.inner
70            .observe_reads(scope, on_value_changed_for_scope, block)
71    }
72
73    /// Notify the observer that a new composition frame is starting.
74    pub fn begin_frame(&self) {
75        self.inner.begin_frame();
76    }
77
78    /// Drop bookkeeping for scopes that were released during the current frame.
79    pub fn prune_dead_scopes(&self) {
80        self.inner.prune_dead_scopes();
81    }
82
83    /// Temporarily pause read observation while executing `block`.
84    pub fn with_no_observations<R>(&self, block: impl FnOnce() -> R) -> R {
85        self.inner.with_no_observations(block)
86    }
87
88    /// Remove any recorded reads for `scope`.
89    pub fn clear<T>(&self, scope: &T)
90    where
91        T: Any + Eq + Hash + 'static,
92    {
93        self.inner.clear(scope);
94    }
95
96    /// Remove recorded reads for scopes that satisfy `predicate`.
97    pub fn clear_if(&self, predicate: impl Fn(&dyn Any) -> bool) {
98        self.inner.clear_if(predicate);
99    }
100
101    /// Remove all recorded observations.
102    pub fn clear_all(&self) {
103        self.inner.clear_all();
104    }
105
106    /// Begin listening for snapshot apply notifications.
107    pub fn start(&self) {
108        let weak = Rc::downgrade(&self.inner);
109        self.inner.start(weak);
110    }
111
112    /// Stop listening for snapshot apply notifications.
113    pub fn stop(&self) {
114        self.inner.stop();
115    }
116
117    pub fn debug_stats(&self) -> SnapshotStateObserverDebugStats {
118        self.inner.debug_stats()
119    }
120
121    /// Test-only helper to simulate snapshot changes.
122    #[cfg(test)]
123    pub fn notify_changes(&self, modified: &[Arc<dyn StateObject>]) {
124        self.inner.handle_apply(modified);
125    }
126}
127
128struct SnapshotStateObserverInner {
129    // Borrowing invariants:
130    // - `observe_reads` may push/pop `active_read_targets`, but it must not hold that
131    //   RefCell borrow across the user-provided `block`; the scoped guard only keeps
132    //   ownership of the target handle, not an active borrow.
133    // - `read_dispatcher` borrows `active_read_targets` just long enough to clone the
134    //   current target, then mutates only the selected `ObservedIds`.
135    // - `handle_apply` snapshots `observed_to_scopes` and `indexed_scopes` into a
136    //   separate notification list before borrowing any `ScopeEntry`, so map borrows
137    //   are never held while callbacks may inspect or mutate entry state.
138    // - `replace_observed_ids`, `clear`, `clear_if`, and `prune_dead_scopes` may borrow
139    //   scope maps mutably, but they must release any `ScopeEntry` borrow before removing
140    //   the entry from `indexed_scopes`, `fast_scopes`, `owned_scopes`, or
141    //   `observed_to_scopes`.
142    executor: Rc<Executor>,
143    owned_scopes: RefCell<HashMap<OwnedScopeIndexKey, OwnedScopeBucket>>,
144    fast_scopes: RefCell<HashMap<ScopeId, Rc<RefCell<ScopeEntry>>>>,
145    indexed_scopes: RefCell<HashMap<usize, Rc<RefCell<ScopeEntry>>>>,
146    observed_to_scopes: RefCell<HashMap<StateObjectId, HashSet<usize>>>,
147    pause_count: Rc<Cell<usize>>,
148    active_read_targets: Rc<RefCell<Vec<Rc<RefCell<ObservedIds>>>>>,
149    read_dispatcher: ReadObserver,
150    apply_handle: RefCell<Option<crate::snapshot_v2::ObserverHandle>>,
151    weak_self: RefCell<Weak<SnapshotStateObserverInner>>,
152    frame_version: Cell<u64>,
153    next_entry_id: Cell<usize>,
154}
155
156#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
157struct OwnedScopeIndexKey {
158    type_id: TypeId,
159    value_hash: u64,
160}
161
162type OwnedScopeBucket = SmallVec<[Rc<RefCell<ScopeEntry>>; 1]>;
163
164fn owned_scope_index_key<T>(scope: &T) -> OwnedScopeIndexKey
165where
166    T: Any + Hash + 'static,
167{
168    let mut hasher = default_hash::new();
169    scope.hash(&mut hasher);
170    OwnedScopeIndexKey {
171        type_id: TypeId::of::<T>(),
172        value_hash: hasher.finish(),
173    }
174}
175
176impl SnapshotStateObserverInner {
177    const MIN_RETAINED_SCOPE_CAPACITY: usize = 256;
178
179    fn new(on_changed_executor: impl Fn(Box<dyn FnOnce() + 'static>) + 'static) -> Self {
180        let pause_count = Rc::new(Cell::new(0));
181        let active_read_targets = Rc::new(RefCell::new(Vec::<Rc<RefCell<ObservedIds>>>::new()));
182        let dispatcher_pause_count = Rc::clone(&pause_count);
183        let dispatcher_targets = Rc::clone(&active_read_targets);
184        let read_dispatcher: ReadObserver = Arc::new(move |state| {
185            if dispatcher_pause_count.get() > 0 {
186                return;
187            }
188            let observed = dispatcher_targets.borrow().last().cloned();
189            if let Some(observed) = observed {
190                observed.borrow_mut().insert(state);
191            }
192        });
193
194        Self {
195            executor: Rc::new(on_changed_executor),
196            owned_scopes: RefCell::new(HashMap::default()),
197            fast_scopes: RefCell::new(HashMap::default()),
198            indexed_scopes: RefCell::new(HashMap::default()),
199            observed_to_scopes: RefCell::new(HashMap::default()),
200            pause_count,
201            active_read_targets,
202            read_dispatcher,
203            apply_handle: RefCell::new(None),
204            weak_self: RefCell::new(Weak::new()),
205            frame_version: Cell::new(0),
206            next_entry_id: Cell::new(0),
207        }
208    }
209
210    fn set_self(&self, weak: Weak<SnapshotStateObserverInner>) {
211        self.weak_self.replace(weak);
212    }
213
214    fn begin_frame(&self) {
215        let next = self.frame_version.get().wrapping_add(1);
216        self.frame_version.set(next);
217        self.prune_dead_scopes();
218    }
219
220    fn observe_reads<T, R>(
221        &self,
222        scope: T,
223        on_value_changed_for_scope: impl Fn(&T) + 'static,
224        block: impl FnOnce() -> R,
225    ) -> R
226    where
227        T: Any + Clone + Eq + Hash + 'static,
228    {
229        let frame_version = self.frame_version.get();
230        let has_frame_version = frame_version != 0;
231
232        let on_changed: Rc<dyn Fn(&dyn Any)> = {
233            let callback = Rc::new(on_value_changed_for_scope);
234            Rc::new(move |scope_any: &dyn Any| {
235                if let Some(typed) = scope_any.downcast_ref::<T>() {
236                    callback(typed);
237                }
238            })
239        };
240
241        let existing_entry = self.find_scope_entry(&scope);
242        if let Some(entry) = existing_entry.as_ref() {
243            let already_observed = {
244                let mut entry_mut = entry.borrow_mut();
245                entry_mut.update(scope.clone(), on_changed.clone());
246                has_frame_version && entry_mut.last_seen_version == frame_version
247            };
248            if already_observed {
249                return block();
250            }
251        }
252
253        let observed = Rc::new(RefCell::new(ObservedIds::new()));
254        self.active_read_targets
255            .borrow_mut()
256            .push(Rc::clone(&observed));
257        struct ActiveObservationGuard {
258            stack: Rc<RefCell<Vec<Rc<RefCell<ObservedIds>>>>>,
259        }
260        impl Drop for ActiveObservationGuard {
261            fn drop(&mut self) {
262                self.stack.borrow_mut().pop();
263            }
264        }
265        let _guard = ActiveObservationGuard {
266            stack: Rc::clone(&self.active_read_targets),
267        };
268
269        let result = self.run_with_read_observer(block);
270
271        if observed.borrow().is_empty() {
272            if existing_entry.is_some() {
273                self.clear(&scope);
274            }
275            return result;
276        }
277
278        let observed = {
279            let mut observed = observed.borrow_mut();
280            std::mem::replace(&mut *observed, ObservedIds::new())
281        };
282        let entry = existing_entry
283            .unwrap_or_else(|| self.insert_scope_entry(scope.clone(), on_changed.clone()));
284        {
285            let mut entry_mut = entry.borrow_mut();
286            entry_mut.update(scope, on_changed);
287            entry_mut.last_seen_version = if has_frame_version {
288                frame_version
289            } else {
290                u64::MAX
291            };
292        }
293        self.replace_observed_ids(&entry, observed);
294
295        result
296    }
297
298    fn with_no_observations<R>(&self, block: impl FnOnce() -> R) -> R {
299        self.pause_count.set(self.pause_count.get() + 1);
300        let result = block();
301        self.pause_count
302            .set(self.pause_count.get().saturating_sub(1));
303        result
304    }
305
306    fn clear<T>(&self, scope: &T)
307    where
308        T: Any + Eq + Hash + 'static,
309    {
310        if let Some(rc_scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
311            if let Some(entry) = self.fast_scopes.borrow_mut().remove(&rc_scope.id()) {
312                self.unregister_entry(&entry);
313            }
314            return;
315        }
316
317        let removed = self.remove_owned_scope_entry(scope);
318        if let Some(entry) = removed {
319            self.unregister_entry(&entry);
320        }
321    }
322
323    fn clear_if(&self, predicate: impl Fn(&dyn Any) -> bool) {
324        let removed_fast = {
325            let mut fast_scopes = self.fast_scopes.borrow_mut();
326            let removed_ids: Vec<_> = fast_scopes
327                .iter()
328                .filter(|(_, entry)| entry.borrow().matches_predicate(&predicate))
329                .map(|(scope_id, _)| *scope_id)
330                .collect();
331            removed_ids
332                .into_iter()
333                .filter_map(|scope_id| fast_scopes.remove(&scope_id))
334                .collect::<Vec<_>>()
335        };
336        let removed_owned =
337            { self.partition_owned_scopes(|entry| entry.matches_predicate(&predicate)) };
338
339        for entry in removed_fast.into_iter().chain(removed_owned) {
340            self.unregister_entry(&entry);
341        }
342    }
343
344    fn clear_all(&self) {
345        self.fast_scopes.borrow_mut().clear();
346        self.owned_scopes.borrow_mut().clear();
347        self.indexed_scopes.borrow_mut().clear();
348        self.observed_to_scopes.borrow_mut().clear();
349    }
350
351    fn start(&self, weak_self: Weak<SnapshotStateObserverInner>) {
352        if self.apply_handle.borrow().is_some() {
353            return;
354        }
355
356        let handle = register_apply_observer(Rc::new(move |modified, _snapshot_id| {
357            if let Some(inner) = weak_self.upgrade() {
358                inner.handle_apply(modified);
359            }
360        }));
361        self.apply_handle.replace(Some(handle));
362    }
363
364    fn stop(&self) {
365        if let Some(handle) = self.apply_handle.borrow_mut().take() {
366            drop(handle);
367        }
368    }
369
370    fn find_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
371    where
372        T: Any + Eq + Hash + 'static,
373    {
374        if let Some(scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
375            return self.fast_scopes.borrow().get(&scope.id()).cloned();
376        }
377
378        self.find_owned_scope_entry(scope)
379    }
380
381    fn insert_scope_entry(
382        &self,
383        scope: impl Any + Clone + Eq + Hash + 'static,
384        on_changed: Rc<dyn Fn(&dyn Any)>,
385    ) -> Rc<RefCell<ScopeEntry>> {
386        let entry_id = self.next_entry_id.get();
387        self.next_entry_id.set(entry_id.wrapping_add(1));
388        let recompose_scope_id = (&scope as &dyn Any)
389            .downcast_ref::<RecomposeScope>()
390            .map(RecomposeScope::id);
391        let owned_scope_key = recompose_scope_id
392            .is_none()
393            .then(|| owned_scope_index_key(&scope));
394        let entry = Rc::new(RefCell::new(ScopeEntry::new(entry_id, scope, on_changed)));
395        self.indexed_scopes
396            .borrow_mut()
397            .insert(entry_id, Rc::clone(&entry));
398        if let Some(scope_id) = recompose_scope_id {
399            self.fast_scopes
400                .borrow_mut()
401                .insert(scope_id, Rc::clone(&entry));
402        } else if let Some(scope_key) = owned_scope_key {
403            self.owned_scopes
404                .borrow_mut()
405                .entry(scope_key)
406                .or_default()
407                .push(Rc::clone(&entry));
408        }
409        entry
410    }
411
412    fn prune_dead_scopes(&self) {
413        let removed_fast = {
414            let mut fast_scopes = self.fast_scopes.borrow_mut();
415            let removed_ids: Vec<_> = fast_scopes
416                .iter()
417                .filter(|(_, entry)| !entry.borrow().should_retain())
418                .map(|(scope_id, _)| *scope_id)
419                .collect();
420            let removed = removed_ids
421                .into_iter()
422                .filter_map(|scope_id| fast_scopes.remove(&scope_id))
423                .collect::<Vec<_>>();
424            shrink_map_if_sparse(&mut fast_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
425            removed
426        };
427
428        let removed_owned = { self.partition_owned_scopes(|entry| !entry.should_retain()) };
429
430        for entry in removed_fast.into_iter().chain(removed_owned) {
431            self.unregister_entry(&entry);
432        }
433    }
434
435    fn find_owned_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
436    where
437        T: Any + Eq + Hash + 'static,
438    {
439        let key = owned_scope_index_key(scope);
440        self.owned_scopes.borrow().get(&key).and_then(|bucket| {
441            bucket
442                .iter()
443                .find(|entry| entry.borrow().matches_scope(scope))
444                .cloned()
445        })
446    }
447
448    fn remove_owned_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
449    where
450        T: Any + Eq + Hash + 'static,
451    {
452        let key = owned_scope_index_key(scope);
453        let mut owned_scopes = self.owned_scopes.borrow_mut();
454        let mut removed = None;
455        let mut remove_bucket = false;
456        if let Some(bucket) = owned_scopes.get_mut(&key) {
457            if let Some(index) = bucket
458                .iter()
459                .position(|entry| entry.borrow().matches_scope(scope))
460            {
461                removed = Some(bucket.remove(index));
462                remove_bucket = bucket.is_empty();
463            }
464        }
465        if remove_bucket {
466            owned_scopes.remove(&key);
467        }
468        shrink_map_if_sparse(&mut owned_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
469        removed
470    }
471
472    fn partition_owned_scopes(
473        &self,
474        should_remove: impl Fn(&ScopeEntry) -> bool,
475    ) -> Vec<Rc<RefCell<ScopeEntry>>> {
476        let mut owned_scopes = self.owned_scopes.borrow_mut();
477        let mut retained = HashMap::default();
478        let mut removed = Vec::new();
479        for (key, mut bucket) in owned_scopes.drain() {
480            let mut retained_bucket = OwnedScopeBucket::new();
481            for entry in bucket.drain(..) {
482                if should_remove(&entry.borrow()) {
483                    removed.push(entry);
484                } else {
485                    retained_bucket.push(entry);
486                }
487            }
488            if !retained_bucket.is_empty() {
489                retained.insert(key, retained_bucket);
490            }
491        }
492        *owned_scopes = retained;
493        shrink_map_if_sparse(&mut owned_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
494        removed
495    }
496
497    fn debug_stats(&self) -> SnapshotStateObserverDebugStats {
498        let owned_scopes = self.owned_scopes.borrow();
499        let fast_scopes = self.fast_scopes.borrow();
500        let owned_scope_len = owned_scopes.values().map(SmallVec::len).sum::<usize>();
501        let owned_scope_cap =
502            owned_scopes.capacity() + owned_scopes.values().map(SmallVec::capacity).sum::<usize>();
503        let scopes_len = owned_scope_len + fast_scopes.len();
504        let scopes_cap = owned_scope_cap + fast_scopes.capacity();
505        let mut observed_state_count = 0;
506        let mut observed_state_capacity = 0;
507        let mut stateless_scope_count = 0;
508
509        for entry in owned_scopes
510            .values()
511            .flat_map(|bucket| bucket.iter())
512            .chain(fast_scopes.values())
513        {
514            let entry = entry.borrow();
515            observed_state_count += entry.observed.len();
516            observed_state_capacity += entry.observed.capacity();
517            stateless_scope_count += usize::from(entry.observed.is_empty());
518        }
519
520        SnapshotStateObserverDebugStats {
521            scopes_len,
522            scopes_cap,
523            fast_scopes_len: fast_scopes.len(),
524            fast_scopes_cap: fast_scopes.capacity(),
525            stateless_scope_count,
526            observed_state_count,
527            observed_state_capacity,
528        }
529    }
530
531    fn run_with_read_observer<R>(&self, block: impl FnOnce() -> R) -> R {
532        // Kotlin uses Snapshot.observeInternal which creates a TransparentObserverMutableSnapshot,
533        // not a readonly snapshot. This allows writes to happen during observation (composition).
534        use crate::snapshot_v2::take_transparent_observer_mutable_snapshot;
535
536        // Create a transparent mutable snapshot (not readonly!) for observation
537        // This matches Kotlin's Snapshot.observeInternal behavior
538        let snapshot =
539            take_transparent_observer_mutable_snapshot(Some(self.read_dispatcher.clone()), None);
540        let result = snapshot.enter(block);
541        snapshot.dispose();
542        result
543    }
544
545    fn handle_apply(&self, modified: &[Arc<dyn StateObject>]) {
546        if modified.is_empty() {
547            return;
548        }
549
550        let mut seen_scope_ids: HashSet<usize> = HashSet::default();
551        let mut to_notify: Vec<Rc<RefCell<ScopeEntry>>> = Vec::new();
552        {
553            let observed_to_scopes = self.observed_to_scopes.borrow();
554            let indexed_scopes = self.indexed_scopes.borrow();
555            for state in modified {
556                if let Some(scope_ids) = observed_to_scopes.get(&state.object_id().as_usize()) {
557                    let mut ordered_scope_ids: SmallVec<[usize; 8]> =
558                        scope_ids.iter().copied().collect();
559                    ordered_scope_ids.sort_unstable();
560                    for scope_id in ordered_scope_ids {
561                        if seen_scope_ids.insert(scope_id) {
562                            if let Some(entry) = indexed_scopes.get(&scope_id) {
563                                to_notify.push(entry.clone());
564                            }
565                        }
566                    }
567                }
568            }
569        }
570
571        if to_notify.is_empty() {
572            return;
573        }
574
575        for entry in to_notify {
576            let executor = self.executor.clone();
577            executor(Box::new(move || {
578                if let Ok(entry) = entry.try_borrow() {
579                    entry.notify();
580                }
581            }));
582        }
583    }
584
585    fn replace_observed_ids(&self, entry: &Rc<RefCell<ScopeEntry>>, observed: ObservedIds) {
586        let (entry_id, previous) = {
587            let mut entry_mut = entry.borrow_mut();
588            let entry_id = entry_mut.id;
589            let previous = std::mem::replace(&mut entry_mut.observed, observed);
590            (entry_id, previous)
591        };
592        self.unregister_observed_ids(entry_id, &previous);
593        let entry_ref = entry.borrow();
594        self.register_observed_ids(entry_id, &entry_ref.observed);
595    }
596
597    fn register_observed_ids(&self, entry_id: usize, observed: &ObservedIds) {
598        let mut observed_to_scopes = self.observed_to_scopes.borrow_mut();
599        for state_id in observed.iter() {
600            let scope_ids = observed_to_scopes.entry(state_id).or_default();
601            scope_ids.insert(entry_id);
602        }
603    }
604
605    fn unregister_observed_ids(&self, entry_id: usize, observed: &ObservedIds) {
606        let mut observed_to_scopes = self.observed_to_scopes.borrow_mut();
607        let mut emptied = SmallVec::<[StateObjectId; MAX_OBSERVED_STATES]>::new();
608        for state_id in observed.iter() {
609            if let Some(scope_ids) = observed_to_scopes.get_mut(&state_id) {
610                scope_ids.remove(&entry_id);
611                if scope_ids.is_empty() {
612                    emptied.push(state_id);
613                }
614            }
615        }
616        for state_id in emptied {
617            observed_to_scopes.remove(&state_id);
618        }
619        shrink_map_if_sparse(&mut observed_to_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
620    }
621
622    fn unregister_entry(&self, entry: &Rc<RefCell<ScopeEntry>>) {
623        let (entry_id, observed) = {
624            let mut entry_mut = entry.borrow_mut();
625            let observed = std::mem::replace(&mut entry_mut.observed, ObservedIds::new());
626            (entry_mut.id, observed)
627        };
628        self.unregister_observed_ids(entry_id, &observed);
629        self.indexed_scopes.borrow_mut().remove(&entry_id);
630    }
631}
632
633fn shrink_map_if_sparse<K, V>(map: &mut HashMap<K, V>, min_retained_capacity: usize)
634where
635    K: Eq + std::hash::Hash,
636{
637    if map.capacity() <= map.len().max(min_retained_capacity).saturating_mul(4) {
638        return;
639    }
640
641    let retained = map.len().max(min_retained_capacity);
642    let mut rebuilt = HashMap::default();
643    rebuilt.reserve(retained);
644    rebuilt.extend(map.drain());
645    *map = rebuilt;
646}
647
648enum ObservedIds {
649    Small(SmallVec<[ObservedState; MAX_OBSERVED_STATES]>),
650    Large(HashMap<StateObjectId, Option<Box<dyn Any>>>),
651}
652
653struct ObservedState {
654    id: StateObjectId,
655    _lease: Option<Box<dyn Any>>,
656}
657
658impl ObservedIds {
659    fn new() -> Self {
660        ObservedIds::Small(SmallVec::new())
661    }
662
663    fn insert(&mut self, state: &dyn StateObject) {
664        let id = state.object_id().as_usize();
665        match self {
666            ObservedIds::Small(small) => {
667                if small.iter().any(|observed| observed.id == id) {
668                    return;
669                }
670                if small.len() < MAX_OBSERVED_STATES {
671                    small.push(ObservedState {
672                        id,
673                        _lease: state.observation_lease(),
674                    });
675                } else {
676                    let mut large =
677                        HashMap::with_capacity_and_hasher(small.len() + 1, Default::default());
678                    for observed in small.drain(..) {
679                        large.insert(observed.id, observed._lease);
680                    }
681                    large.insert(id, state.observation_lease());
682                    *self = ObservedIds::Large(large);
683                }
684            }
685            ObservedIds::Large(large) => {
686                if !large.contains_key(&id) {
687                    large.insert(id, state.observation_lease());
688                }
689            }
690        }
691    }
692
693    fn is_empty(&self) -> bool {
694        match self {
695            ObservedIds::Small(small) => small.is_empty(),
696            ObservedIds::Large(large) => large.is_empty(),
697        }
698    }
699
700    fn len(&self) -> usize {
701        match self {
702            ObservedIds::Small(small) => small.len(),
703            ObservedIds::Large(large) => large.len(),
704        }
705    }
706
707    fn capacity(&self) -> usize {
708        match self {
709            ObservedIds::Small(small) => small.capacity(),
710            ObservedIds::Large(large) => large.capacity(),
711        }
712    }
713
714    fn iter(&self) -> Box<dyn Iterator<Item = StateObjectId> + '_> {
715        match self {
716            ObservedIds::Small(small) => Box::new(small.iter().map(|observed| observed.id)),
717            ObservedIds::Large(large) => Box::new(large.keys().copied()),
718        }
719    }
720}
721
722const MAX_OBSERVED_STATES: usize = 8;
723
724enum ScopeStorage {
725    Owned(Box<dyn Any>),
726    RecomposeScope {
727        id: ScopeId,
728        weak: Weak<RecomposeScopeInner>,
729    },
730}
731
732struct ScopeEntry {
733    id: usize,
734    scope: ScopeStorage,
735    on_changed: Rc<dyn Fn(&dyn Any)>,
736    observed: ObservedIds,
737    last_seen_version: u64,
738}
739
740impl ScopeEntry {
741    fn new<T>(id: usize, scope: T, on_changed: Rc<dyn Fn(&dyn Any)>) -> Self
742    where
743        T: Any + 'static,
744    {
745        Self {
746            id,
747            scope: ScopeStorage::from_value(scope),
748            on_changed,
749            observed: ObservedIds::new(),
750            last_seen_version: u64::MAX,
751        }
752    }
753
754    fn update<T>(&mut self, new_scope: T, on_changed: Rc<dyn Fn(&dyn Any)>)
755    where
756        T: Any + 'static,
757    {
758        self.scope = ScopeStorage::from_value(new_scope);
759        self.on_changed = on_changed;
760    }
761
762    fn matches_scope<T>(&self, scope: &T) -> bool
763    where
764        T: Any + Eq + 'static,
765    {
766        if let Some(scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
767            return matches!(
768                &self.scope,
769                ScopeStorage::RecomposeScope { id, .. } if *id == scope.id()
770            );
771        }
772
773        match &self.scope {
774            ScopeStorage::Owned(stored) => stored
775                .downcast_ref::<T>()
776                .map(|stored| stored == scope)
777                .unwrap_or(false),
778            ScopeStorage::RecomposeScope { .. } => false,
779        }
780    }
781
782    fn matches_predicate(&self, predicate: &impl Fn(&dyn Any) -> bool) -> bool {
783        match &self.scope {
784            ScopeStorage::Owned(scope) => predicate(scope.as_ref()),
785            ScopeStorage::RecomposeScope { weak, .. } => weak
786                .upgrade()
787                .map(|inner| predicate(&RecomposeScope { inner }))
788                .unwrap_or(true),
789        }
790    }
791
792    fn should_retain(&self) -> bool {
793        match &self.scope {
794            ScopeStorage::Owned(_) => true,
795            ScopeStorage::RecomposeScope { weak, .. } => weak.upgrade().is_some(),
796        }
797    }
798
799    fn notify(&self) {
800        match &self.scope {
801            ScopeStorage::Owned(scope) => (self.on_changed)(scope.as_ref()),
802            ScopeStorage::RecomposeScope { weak, .. } => {
803                if let Some(inner) = weak.upgrade() {
804                    (self.on_changed)(&RecomposeScope { inner });
805                }
806            }
807        }
808    }
809}
810
811impl ScopeStorage {
812    fn from_value<T>(value: T) -> Self
813    where
814        T: Any + 'static,
815    {
816        let any = &value as &dyn Any;
817        if let Some(scope) = any.downcast_ref::<RecomposeScope>() {
818            Self::RecomposeScope {
819                id: scope.id(),
820                weak: scope.downgrade(),
821            }
822        } else {
823            Self::Owned(Box::new(value))
824        }
825    }
826}
827
828#[cfg(test)]
829mod tests {
830    use super::*;
831    use crate::snapshot_v2::take_mutable_snapshot;
832    use crate::snapshot_v2::{reset_runtime_for_tests, TestRuntimeGuard};
833    use crate::state::{NeverEqual, SnapshotMutableState};
834    use std::cell::Cell;
835
836    fn reset_runtime() -> TestRuntimeGuard {
837        reset_runtime_for_tests()
838    }
839
840    #[derive(Clone, Eq, Hash, PartialEq)]
841    struct TestScope(&'static str);
842
843    #[test]
844    fn notifies_scope_when_state_changes() {
845        let _guard = reset_runtime();
846
847        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
848        let triggered = Rc::new(Cell::new(0));
849        let observer_trigger = triggered.clone();
850
851        let observer = SnapshotStateObserver::new(|callback| callback());
852        observer.start();
853
854        let scope = TestScope("scope");
855        observer.observe_reads(
856            scope.clone(),
857            move |_| {
858                observer_trigger.set(observer_trigger.get() + 1);
859            },
860            || {
861                let _ = state.get();
862            },
863        );
864
865        let snapshot = take_mutable_snapshot(None, None);
866        snapshot.enter(|| {
867            state.set(1);
868        });
869        snapshot.apply().check();
870
871        assert_eq!(triggered.get(), 1);
872        observer.stop();
873    }
874
875    #[test]
876    fn clear_removes_scope_observation() {
877        let _guard = reset_runtime();
878
879        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
880        let triggered = Rc::new(Cell::new(0));
881        let observer_trigger = triggered.clone();
882
883        let observer = SnapshotStateObserver::new(|callback| callback());
884        observer.start();
885
886        let scope = TestScope("scope");
887        observer.observe_reads(
888            scope.clone(),
889            move |_| {
890                observer_trigger.set(observer_trigger.get() + 1);
891            },
892            || {
893                let _ = state.get();
894            },
895        );
896
897        observer.clear(&scope);
898
899        let snapshot = take_mutable_snapshot(None, None);
900        snapshot.enter(|| {
901            state.set(1);
902        });
903        snapshot.apply().check();
904
905        assert_eq!(triggered.get(), 0);
906        observer.stop();
907    }
908
909    #[test]
910    fn repeated_owned_scope_observations_reuse_the_same_entry() {
911        let _guard = reset_runtime();
912
913        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
914        let observer = SnapshotStateObserver::new(|callback| callback());
915        let scope = TestScope("scope");
916
917        observer.observe_reads(
918            scope.clone(),
919            |_| {},
920            || {
921                let _ = state.get();
922            },
923        );
924        observer.observe_reads(
925            scope,
926            |_| {},
927            || {
928                let _ = state.get();
929            },
930        );
931
932        let stats = observer.debug_stats();
933        assert_eq!(stats.scopes_len, 1);
934        assert_eq!(stats.fast_scopes_len, 0);
935    }
936
937    #[test]
938    fn with_no_observations_skips_reads() {
939        let _guard = reset_runtime();
940
941        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
942        let triggered = Rc::new(Cell::new(0));
943        let observer_trigger = triggered.clone();
944
945        let observer = SnapshotStateObserver::new(|callback| callback());
946        observer.start();
947
948        let scope = TestScope("scope");
949        observer.observe_reads(
950            scope.clone(),
951            move |_| {
952                observer_trigger.set(observer_trigger.get() + 1);
953            },
954            || {
955                observer.with_no_observations(|| {
956                    let _ = state.get();
957                });
958            },
959        );
960
961        let snapshot = take_mutable_snapshot(None, None);
962        snapshot.enter(|| {
963            state.set(1);
964        });
965        snapshot.apply().check();
966
967        assert_eq!(triggered.get(), 0);
968        observer.stop();
969    }
970
971    #[test]
972    fn nested_observe_reads_attributes_state_to_innermost_scope_only() {
973        let _guard = reset_runtime();
974
975        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
976        let outer_triggered = Rc::new(Cell::new(0));
977        let inner_triggered = Rc::new(Cell::new(0));
978
979        let observer = SnapshotStateObserver::new(|callback| callback());
980        observer.start();
981
982        let outer_scope = TestScope("outer");
983        let inner_scope = TestScope("inner");
984        observer.observe_reads(
985            outer_scope.clone(),
986            {
987                let outer_triggered = Rc::clone(&outer_triggered);
988                move |_| outer_triggered.set(outer_triggered.get() + 1)
989            },
990            || {
991                observer.observe_reads(
992                    inner_scope.clone(),
993                    {
994                        let inner_triggered = Rc::clone(&inner_triggered);
995                        move |_| inner_triggered.set(inner_triggered.get() + 1)
996                    },
997                    || {
998                        let _ = state.get();
999                    },
1000                );
1001            },
1002        );
1003
1004        let snapshot = take_mutable_snapshot(None, None);
1005        snapshot.enter(|| {
1006            state.set(1);
1007        });
1008        snapshot.apply().check();
1009
1010        assert_eq!(outer_triggered.get(), 0);
1011        assert_eq!(inner_triggered.get(), 1);
1012        observer.stop();
1013    }
1014
1015    #[test]
1016    fn clearing_one_scope_keeps_shared_state_registered_for_other_scope() {
1017        let _guard = reset_runtime();
1018
1019        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1020        let first_triggered = Rc::new(Cell::new(0));
1021        let second_triggered = Rc::new(Cell::new(0));
1022
1023        let observer = SnapshotStateObserver::new(|callback| callback());
1024        observer.start();
1025
1026        let first_scope = TestScope("first");
1027        let second_scope = TestScope("second");
1028        observer.observe_reads(
1029            first_scope.clone(),
1030            {
1031                let first_triggered = Rc::clone(&first_triggered);
1032                move |_| first_triggered.set(first_triggered.get() + 1)
1033            },
1034            || {
1035                let _ = state.get();
1036            },
1037        );
1038        observer.observe_reads(
1039            second_scope.clone(),
1040            {
1041                let second_triggered = Rc::clone(&second_triggered);
1042                move |_| second_triggered.set(second_triggered.get() + 1)
1043            },
1044            || {
1045                let _ = state.get();
1046            },
1047        );
1048
1049        observer.clear(&first_scope);
1050
1051        let snapshot = take_mutable_snapshot(None, None);
1052        snapshot.enter(|| {
1053            state.set(1);
1054        });
1055        snapshot.apply().check();
1056
1057        assert_eq!(first_triggered.get(), 0);
1058        assert_eq!(second_triggered.get(), 1);
1059        observer.stop();
1060    }
1061
1062    #[test]
1063    fn shared_state_notifies_scopes_in_registration_order() {
1064        let _guard = reset_runtime();
1065
1066        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1067        let notifications = Rc::new(RefCell::new(Vec::new()));
1068
1069        let observer = SnapshotStateObserver::new(|callback| callback());
1070        observer.start();
1071
1072        observer.observe_reads(
1073            TestScope("first"),
1074            {
1075                let notifications = Rc::clone(&notifications);
1076                move |_| notifications.borrow_mut().push("first")
1077            },
1078            || {
1079                let _ = state.get();
1080            },
1081        );
1082        observer.observe_reads(
1083            TestScope("second"),
1084            {
1085                let notifications = Rc::clone(&notifications);
1086                move |_| notifications.borrow_mut().push("second")
1087            },
1088            || {
1089                let _ = state.get();
1090            },
1091        );
1092
1093        let snapshot = take_mutable_snapshot(None, None);
1094        snapshot.enter(|| {
1095            state.set(1);
1096        });
1097        snapshot.apply().check();
1098
1099        assert_eq!(notifications.borrow().as_slice(), &["first", "second"]);
1100        observer.stop();
1101    }
1102
1103    #[test]
1104    fn stateless_recompose_scope_does_not_retain_observer_entry() {
1105        let _guard = reset_runtime();
1106
1107        let observer = SnapshotStateObserver::new(|callback| callback());
1108        let runtime = crate::TestRuntime::new();
1109        let scope = RecomposeScope::new_for_test(runtime.handle());
1110
1111        observer.observe_reads(scope, |_| {}, || {});
1112
1113        let stats = observer.debug_stats();
1114        assert_eq!(stats.scopes_len, 0);
1115        assert_eq!(stats.fast_scopes_len, 0);
1116        assert_eq!(stats.stateless_scope_count, 0);
1117    }
1118
1119    #[test]
1120    fn scope_that_stops_reading_state_is_removed_immediately() {
1121        let _guard = reset_runtime();
1122
1123        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1124        let observer = SnapshotStateObserver::new(|callback| callback());
1125        let runtime = crate::TestRuntime::new();
1126        let scope = RecomposeScope::new_for_test(runtime.handle());
1127        let triggered = Rc::new(Cell::new(0));
1128        let observer_trigger = Rc::clone(&triggered);
1129
1130        observer.observe_reads(
1131            scope.clone(),
1132            move |_| observer_trigger.set(observer_trigger.get() + 1),
1133            || {
1134                let _ = state.get();
1135            },
1136        );
1137
1138        let after_stateful = observer.debug_stats();
1139        assert_eq!(after_stateful.scopes_len, 1);
1140        assert_eq!(after_stateful.fast_scopes_len, 1);
1141
1142        observer.observe_reads(scope, |_| {}, || {});
1143
1144        let after_stateless = observer.debug_stats();
1145        assert_eq!(after_stateless.scopes_len, 0);
1146        assert_eq!(after_stateless.fast_scopes_len, 0);
1147
1148        let snapshot = take_mutable_snapshot(None, None);
1149        snapshot.enter(|| {
1150            state.set(1);
1151        });
1152        snapshot.apply().check();
1153
1154        assert_eq!(triggered.get(), 0);
1155    }
1156
1157    #[test]
1158    fn begin_frame_prunes_dropped_recompose_scope_entries() {
1159        let _guard = reset_runtime();
1160
1161        let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1162        let observer = SnapshotStateObserver::new(|callback| callback());
1163        let runtime = crate::TestRuntime::new();
1164        let scope = RecomposeScope::new_for_test(runtime.handle());
1165
1166        observer.observe_reads(
1167            scope.clone(),
1168            |_| {},
1169            || {
1170                let _ = state.get();
1171            },
1172        );
1173
1174        let before_prune = observer.debug_stats();
1175        assert_eq!(before_prune.scopes_len, 1);
1176        assert_eq!(before_prune.fast_scopes_len, 1);
1177
1178        drop(scope);
1179        observer.begin_frame();
1180
1181        let after_prune = observer.debug_stats();
1182        assert_eq!(after_prune.scopes_len, 0);
1183        assert_eq!(after_prune.fast_scopes_len, 0);
1184    }
1185}