Skip to main content

cranpose_core/snapshot_v2/
mod.rs

1//! Snapshot system for managing isolated state changes.
2//!
3//! This module implements Jetpack Compose's snapshot isolation system, allowing
4//! state changes to be isolated, composed, and atomically applied.
5//!
6//! # Snapshot Types
7//!
8//! - **ReadonlySnapshot**: Immutable view of state at a point in time
9//! - **MutableSnapshot**: Allows isolated state mutations
10//! - **NestedReadonlySnapshot**: Readonly snapshot nested in a parent
11//! - **NestedMutableSnapshot**: Mutable snapshot nested in a parent
12//! - **GlobalSnapshot**: Special global mutable snapshot
13//! - **TransparentObserverMutableSnapshot**: Optimized for observer chaining
14//! - **TransparentObserverSnapshot**: Readonly version of transparent observer
15//!
16//! # Thread Local Storage
17//!
18//! The current snapshot is stored in thread-local storage and automatically
19//! managed by the snapshot system.
20
21#![allow(clippy::arc_with_non_send_sync)]
22
23use std::{
24    cell::{Cell, RefCell},
25    hash::{Hash, Hasher},
26    rc::Rc,
27    sync::{Arc, Weak},
28};
29
30use crate::{
31    collections::map::{HashMap, HashSet},
32    snapshot_id_set::{SnapshotId, SnapshotIdSet},
33    snapshot_pinning::{self, PinHandle},
34    snapshot_weak_set::SnapshotWeakSetDebugStats,
35    state::{StateObject, StateRecord},
36};
37
38mod global;
39mod mutable;
40mod nested;
41mod readonly;
42mod runtime;
43mod transparent;
44
45#[cfg(test)]
46mod integration_tests;
47
48pub use global::{GlobalSnapshot, advance_global_snapshot};
49pub use mutable::MutableSnapshot;
50pub use nested::{NestedMutableSnapshot, NestedReadonlySnapshot};
51pub use readonly::ReadonlySnapshot;
52#[cfg(test)]
53pub(crate) use runtime::{TestRuntimeGuard, reset_runtime_for_tests};
54pub(crate) use runtime::{allocate_snapshot, close_snapshot, with_runtime};
55pub use transparent::{TransparentObserverMutableSnapshot, TransparentObserverSnapshot};
56
57/// Observer that is called when a state object is read.
58pub type ReadObserver = Arc<dyn Fn(&dyn StateObject) + 'static>;
59
60/// Observer that is called when a state object is written.
61pub type WriteObserver = Arc<dyn Fn(&dyn StateObject) + 'static>;
62
63/// Apply observer that is called when a snapshot is applied.
64pub type ApplyObserver = Rc<dyn Fn(&[Arc<dyn StateObject>], SnapshotId) + 'static>;
65
66/// Result of applying a mutable snapshot.
67#[derive(Debug, Clone, PartialEq, Eq)]
68pub enum SnapshotApplyResult {
69    /// The snapshot was applied successfully.
70    Success,
71    /// The snapshot could not be applied due to conflicts.
72    Failure,
73}
74
75impl SnapshotApplyResult {
76    /// Check if the result is successful.
77    pub fn is_success(&self) -> bool {
78        matches!(self, SnapshotApplyResult::Success)
79    }
80
81    /// Check if the result is a failure.
82    pub fn is_failure(&self) -> bool {
83        matches!(self, SnapshotApplyResult::Failure)
84    }
85
86    /// Panic if the result is a failure (for use in tests).
87    #[track_caller]
88    pub fn check(&self) {
89        if self.is_failure() {
90            panic!("Snapshot apply failed");
91        }
92    }
93}
94
95/// Unique identifier for a state object in the modified set.
96pub type StateObjectId = usize;
97
98/// Enum wrapper for all snapshot types.
99///
100/// This provides a type-safe way to work with different snapshot types
101/// without requiring trait objects, which avoids object-safety issues.
102#[derive(Clone)]
103pub enum AnySnapshot {
104    Readonly(Arc<ReadonlySnapshot>),
105    Mutable(Arc<MutableSnapshot>),
106    NestedReadonly(Arc<NestedReadonlySnapshot>),
107    NestedMutable(Arc<NestedMutableSnapshot>),
108    Global(Arc<GlobalSnapshot>),
109    TransparentMutable(Arc<TransparentObserverMutableSnapshot>),
110    TransparentReadonly(Arc<TransparentObserverSnapshot>),
111}
112
113/// Enum wrapper for mutable snapshot types.
114///
115/// This allows `take_mutable_snapshot` to return either a root MutableSnapshot
116/// or a NestedMutableSnapshot depending on the current context, matching Kotlin's
117/// behavior where `takeMutableSnapshot` creates nested snapshots when inside a
118/// mutable snapshot.
119#[derive(Clone)]
120pub enum AnyMutableSnapshot {
121    Root(Arc<MutableSnapshot>),
122    Nested(Arc<NestedMutableSnapshot>),
123}
124
125impl AnyMutableSnapshot {
126    /// Get the snapshot ID.
127    pub fn snapshot_id(&self) -> SnapshotId {
128        match self {
129            AnyMutableSnapshot::Root(s) => s.snapshot_id(),
130            AnyMutableSnapshot::Nested(s) => s.snapshot_id(),
131        }
132    }
133
134    /// Get the set of invalid snapshot IDs.
135    pub fn invalid(&self) -> SnapshotIdSet {
136        match self {
137            AnyMutableSnapshot::Root(s) => s.invalid(),
138            AnyMutableSnapshot::Nested(s) => s.invalid(),
139        }
140    }
141
142    /// Enter this snapshot, making it current for the duration of the closure.
143    pub fn enter<T>(&self, f: impl FnOnce() -> T) -> T {
144        match self {
145            AnyMutableSnapshot::Root(s) => s.enter(f),
146            AnyMutableSnapshot::Nested(s) => s.enter(f),
147        }
148    }
149
150    /// Apply the snapshot.
151    pub fn apply(&self) -> SnapshotApplyResult {
152        match self {
153            AnyMutableSnapshot::Root(s) => s.apply(),
154            AnyMutableSnapshot::Nested(s) => s.apply(),
155        }
156    }
157
158    /// Dispose the snapshot.
159    pub fn dispose(&self) {
160        match self {
161            AnyMutableSnapshot::Root(s) => s.dispose(),
162            AnyMutableSnapshot::Nested(s) => s.dispose(),
163        }
164    }
165}
166
167impl AnySnapshot {
168    /// Get the snapshot ID.
169    pub fn snapshot_id(&self) -> SnapshotId {
170        match self {
171            AnySnapshot::Readonly(s) => s.snapshot_id(),
172            AnySnapshot::Mutable(s) => s.snapshot_id(),
173            AnySnapshot::NestedReadonly(s) => s.snapshot_id(),
174            AnySnapshot::NestedMutable(s) => s.snapshot_id(),
175            AnySnapshot::Global(s) => s.snapshot_id(),
176            AnySnapshot::TransparentMutable(s) => s.snapshot_id(),
177            AnySnapshot::TransparentReadonly(s) => s.snapshot_id(),
178        }
179    }
180
181    /// Get the set of invalid snapshot IDs.
182    pub fn invalid(&self) -> SnapshotIdSet {
183        match self {
184            AnySnapshot::Readonly(s) => s.invalid(),
185            AnySnapshot::Mutable(s) => s.invalid(),
186            AnySnapshot::NestedReadonly(s) => s.invalid(),
187            AnySnapshot::NestedMutable(s) => s.invalid(),
188            AnySnapshot::Global(s) => s.invalid(),
189            AnySnapshot::TransparentMutable(s) => s.invalid(),
190            AnySnapshot::TransparentReadonly(s) => s.invalid(),
191        }
192    }
193
194    /// Check if a snapshot ID is valid in this snapshot.
195    pub fn is_valid(&self, id: SnapshotId) -> bool {
196        let snapshot_id = self.snapshot_id();
197        id <= snapshot_id && !self.invalid().get(id)
198    }
199
200    /// Check if this is a read-only snapshot.
201    pub fn read_only(&self) -> bool {
202        match self {
203            AnySnapshot::Readonly(_) => true,
204            AnySnapshot::Mutable(_) => false,
205            AnySnapshot::NestedReadonly(_) => true,
206            AnySnapshot::NestedMutable(_) => false,
207            AnySnapshot::Global(_) => false,
208            AnySnapshot::TransparentMutable(_) => false,
209            AnySnapshot::TransparentReadonly(_) => true,
210        }
211    }
212
213    /// Get the root snapshot.
214    pub fn root(&self) -> AnySnapshot {
215        match self {
216            AnySnapshot::Readonly(s) => AnySnapshot::Readonly(s.root_readonly()),
217            AnySnapshot::Mutable(s) => AnySnapshot::Mutable(s.root_mutable()),
218            AnySnapshot::NestedReadonly(s) => AnySnapshot::NestedReadonly(s.root_nested_readonly()),
219            AnySnapshot::NestedMutable(s) => AnySnapshot::Mutable(s.root_mutable()),
220            AnySnapshot::Global(s) => AnySnapshot::Global(s.root_global()),
221            AnySnapshot::TransparentMutable(s) => {
222                AnySnapshot::TransparentMutable(s.root_transparent_mutable())
223            }
224            AnySnapshot::TransparentReadonly(s) => {
225                AnySnapshot::TransparentReadonly(s.root_transparent_readonly())
226            }
227        }
228    }
229
230    /// Check if this snapshot refers to the same transparent snapshot.
231    pub fn is_same_transparent(&self, other: &Arc<TransparentObserverMutableSnapshot>) -> bool {
232        matches!(self, AnySnapshot::TransparentMutable(snapshot) if Arc::ptr_eq(snapshot, other))
233    }
234
235    /// Check if this snapshot refers to the same transparent mutable snapshot.
236    pub fn is_same_transparent_mutable(
237        &self,
238        other: &Arc<TransparentObserverMutableSnapshot>,
239    ) -> bool {
240        self.is_same_transparent(other)
241    }
242
243    /// Check if this snapshot refers to the same transparent readonly snapshot.
244    pub fn is_same_transparent_readonly(&self, other: &Arc<TransparentObserverSnapshot>) -> bool {
245        matches!(self, AnySnapshot::TransparentReadonly(snapshot) if Arc::ptr_eq(snapshot, other))
246    }
247
248    /// Enter this snapshot, making it current for the duration of the closure.
249    pub fn enter<T>(&self, f: impl FnOnce() -> T) -> T {
250        match self {
251            AnySnapshot::Readonly(s) => s.enter(f),
252            AnySnapshot::Mutable(s) => s.enter(f),
253            AnySnapshot::NestedReadonly(s) => s.enter(f),
254            AnySnapshot::NestedMutable(s) => s.enter(f),
255            AnySnapshot::Global(s) => s.enter(f),
256            AnySnapshot::TransparentMutable(s) => s.enter(f),
257            AnySnapshot::TransparentReadonly(s) => s.enter(f),
258        }
259    }
260
261    /// Take a nested read-only snapshot.
262    pub fn take_nested_snapshot(&self, read_observer: Option<ReadObserver>) -> AnySnapshot {
263        match self {
264            AnySnapshot::Readonly(s) => {
265                AnySnapshot::Readonly(s.take_nested_snapshot(read_observer))
266            }
267            AnySnapshot::Mutable(s) => AnySnapshot::Readonly(s.take_nested_snapshot(read_observer)),
268            AnySnapshot::NestedReadonly(s) => {
269                AnySnapshot::NestedReadonly(s.take_nested_snapshot(read_observer))
270            }
271            AnySnapshot::NestedMutable(s) => {
272                AnySnapshot::Readonly(s.take_nested_snapshot(read_observer))
273            }
274            AnySnapshot::Global(s) => AnySnapshot::Readonly(s.take_nested_snapshot(read_observer)),
275            AnySnapshot::TransparentMutable(s) => {
276                AnySnapshot::Readonly(s.take_nested_snapshot(read_observer))
277            }
278            AnySnapshot::TransparentReadonly(s) => {
279                AnySnapshot::TransparentReadonly(s.take_nested_snapshot(read_observer))
280            }
281        }
282    }
283
284    /// Check if there are pending changes.
285    pub fn has_pending_changes(&self) -> bool {
286        match self {
287            AnySnapshot::Readonly(s) => s.has_pending_changes(),
288            AnySnapshot::Mutable(s) => s.has_pending_changes(),
289            AnySnapshot::NestedReadonly(s) => s.has_pending_changes(),
290            AnySnapshot::NestedMutable(s) => s.has_pending_changes(),
291            AnySnapshot::Global(s) => s.has_pending_changes(),
292            AnySnapshot::TransparentMutable(s) => s.has_pending_changes(),
293            AnySnapshot::TransparentReadonly(s) => s.has_pending_changes(),
294        }
295    }
296
297    /// Dispose of this snapshot.
298    pub fn dispose(&self) {
299        match self {
300            AnySnapshot::Readonly(s) => s.dispose(),
301            AnySnapshot::Mutable(s) => s.dispose(),
302            AnySnapshot::NestedReadonly(s) => s.dispose(),
303            AnySnapshot::NestedMutable(s) => s.dispose(),
304            AnySnapshot::Global(s) => s.dispose(),
305            AnySnapshot::TransparentMutable(s) => s.dispose(),
306            AnySnapshot::TransparentReadonly(s) => s.dispose(),
307        }
308    }
309
310    /// Check if disposed.
311    pub fn is_disposed(&self) -> bool {
312        match self {
313            AnySnapshot::Readonly(s) => s.is_disposed(),
314            AnySnapshot::Mutable(s) => s.is_disposed(),
315            AnySnapshot::NestedReadonly(s) => s.is_disposed(),
316            AnySnapshot::NestedMutable(s) => s.is_disposed(),
317            AnySnapshot::Global(s) => s.is_disposed(),
318            AnySnapshot::TransparentMutable(s) => s.is_disposed(),
319            AnySnapshot::TransparentReadonly(s) => s.is_disposed(),
320        }
321    }
322
323    /// Record a read.
324    pub fn record_read(&self, state: &dyn StateObject) {
325        match self {
326            AnySnapshot::Readonly(s) => s.record_read(state),
327            AnySnapshot::Mutable(s) => s.record_read(state),
328            AnySnapshot::NestedReadonly(s) => s.record_read(state),
329            AnySnapshot::NestedMutable(s) => s.record_read(state),
330            AnySnapshot::Global(s) => s.record_read(state),
331            AnySnapshot::TransparentMutable(s) => s.record_read(state),
332            AnySnapshot::TransparentReadonly(s) => s.record_read(state),
333        }
334    }
335
336    /// Record a write.
337    pub fn record_write(&self, state: Arc<dyn StateObject>) {
338        match self {
339            AnySnapshot::Readonly(s) => s.record_write(state),
340            AnySnapshot::Mutable(s) => s.record_write(state),
341            AnySnapshot::NestedReadonly(s) => s.record_write(state),
342            AnySnapshot::NestedMutable(s) => s.record_write(state),
343            AnySnapshot::Global(s) => s.record_write(state),
344            AnySnapshot::TransparentMutable(s) => s.record_write(state),
345            AnySnapshot::TransparentReadonly(s) => s.record_write(state),
346        }
347    }
348
349    /// Apply changes (only valid for mutable snapshots).
350    pub fn apply(&self) -> SnapshotApplyResult {
351        match self {
352            AnySnapshot::Mutable(s) => s.apply(),
353            AnySnapshot::NestedMutable(s) => s.apply(),
354            AnySnapshot::Global(s) => s.apply(),
355            AnySnapshot::TransparentMutable(s) => s.apply(),
356            _ => panic!("Cannot apply a read-only snapshot"),
357        }
358    }
359
360    /// Take a nested mutable snapshot (only valid for mutable snapshots).
361    pub fn take_nested_mutable_snapshot(
362        &self,
363        read_observer: Option<ReadObserver>,
364        write_observer: Option<WriteObserver>,
365    ) -> AnySnapshot {
366        match self {
367            AnySnapshot::Mutable(s) => AnySnapshot::NestedMutable(
368                s.take_nested_mutable_snapshot(read_observer, write_observer),
369            ),
370            AnySnapshot::NestedMutable(s) => AnySnapshot::NestedMutable(
371                s.take_nested_mutable_snapshot(read_observer, write_observer),
372            ),
373            AnySnapshot::Global(s) => {
374                AnySnapshot::Mutable(s.take_nested_mutable_snapshot(read_observer, write_observer))
375            }
376            AnySnapshot::TransparentMutable(s) => AnySnapshot::TransparentMutable(
377                s.take_nested_mutable_snapshot(read_observer, write_observer),
378            ),
379            _ => panic!("Cannot take nested mutable snapshot from read-only snapshot"),
380        }
381    }
382}
383
384thread_local! {
385    static CURRENT_SNAPSHOT: RefCell<Option<AnySnapshot>> = const { RefCell::new(None) };
386}
387
388/// Get the current snapshot, or None if not in a snapshot context.
389pub fn current_snapshot() -> Option<AnySnapshot> {
390    CURRENT_SNAPSHOT
391        .try_with(|cell| cell.borrow().clone())
392        .unwrap_or(None)
393}
394
395pub(crate) fn set_current_snapshot(snapshot: Option<AnySnapshot>) {
396    let _ = CURRENT_SNAPSHOT.try_with(|cell| {
397        *cell.borrow_mut() = snapshot;
398    });
399}
400
401struct CurrentSnapshotGuard {
402    previous: Option<AnySnapshot>,
403}
404
405impl CurrentSnapshotGuard {
406    fn enter(snapshot: AnySnapshot) -> Self {
407        let previous = current_snapshot();
408        set_current_snapshot(Some(snapshot));
409        Self { previous }
410    }
411}
412
413impl Drop for CurrentSnapshotGuard {
414    fn drop(&mut self) {
415        set_current_snapshot(self.previous.take());
416    }
417}
418
419pub(crate) fn enter_snapshot_scope<T>(snapshot: AnySnapshot, f: impl FnOnce() -> T) -> T {
420    let _guard = CurrentSnapshotGuard::enter(snapshot);
421    f()
422}
423
424/// Creates a mutable snapshot, matching Kotlin's `Snapshot.takeMutableSnapshot` semantics.
425///
426/// If called while inside a MutableSnapshot, creates a nested snapshot that will
427/// apply to the parent when `apply()` is called. This ensures proper isolation
428/// between nested operations (like event handlers during animations).
429///
430/// If called while inside a GlobalSnapshot or no snapshot, creates a root
431/// mutable snapshot that applies to the global state.
432pub fn take_mutable_snapshot(
433    read_observer: Option<ReadObserver>,
434    write_observer: Option<WriteObserver>,
435) -> AnyMutableSnapshot {
436    match current_snapshot() {
437        Some(AnySnapshot::Mutable(parent)) => AnyMutableSnapshot::Nested(
438            parent.take_nested_mutable_snapshot(read_observer, write_observer),
439        ),
440        Some(AnySnapshot::NestedMutable(parent)) => AnyMutableSnapshot::Nested(
441            parent.take_nested_mutable_snapshot(read_observer, write_observer),
442        ),
443        _ => AnyMutableSnapshot::Root(
444            GlobalSnapshot::get_or_create()
445                .take_nested_mutable_snapshot(read_observer, write_observer),
446        ),
447    }
448}
449
450/// Take a transparent observer mutable snapshot with optional observers.
451///
452/// This type of snapshot is used for read observation during composition,
453/// matching Kotlin's Snapshot.observeInternal behavior. It allows writes
454/// to happen during observation.
455///
456/// Transparent snapshots DO NOT allocate new IDs - they delegate to the
457/// current/global snapshot, making them "transparent" to the snapshot system.
458pub fn take_transparent_observer_mutable_snapshot(
459    read_observer: Option<ReadObserver>,
460    write_observer: Option<WriteObserver>,
461) -> Arc<TransparentObserverMutableSnapshot> {
462    take_transparent_observer_mutable_snapshot_reusing(read_observer, write_observer, None)
463}
464
465pub(crate) fn take_transparent_observer_mutable_snapshot_reusing(
466    read_observer: Option<ReadObserver>,
467    write_observer: Option<WriteObserver>,
468    recycled: Option<Arc<TransparentObserverMutableSnapshot>>,
469) -> Arc<TransparentObserverMutableSnapshot> {
470    let parent = current_snapshot();
471    match parent {
472        Some(AnySnapshot::TransparentMutable(transparent)) if transparent.can_reuse() => {
473            transparent
474        }
475        _ => {
476            let current = current_snapshot()
477                .unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()));
478            let id = current.snapshot_id();
479            let invalid = current.invalid();
480            TransparentObserverMutableSnapshot::new_reusing(
481                recycled,
482                id,
483                invalid,
484                read_observer,
485                write_observer,
486                None,
487            )
488        }
489    }
490}
491
492/// Allocate a new record identifier that is distinct from any active snapshot id.
493pub fn allocate_record_id() -> SnapshotId {
494    runtime::allocate_record_id()
495}
496
497pub(crate) fn peek_next_snapshot_id() -> SnapshotId {
498    runtime::peek_next_snapshot_id()
499}
500
501#[derive(Clone)]
502struct ObserverId(Rc<()>);
503
504impl ObserverId {
505    fn new() -> Self {
506        Self(Rc::new(()))
507    }
508}
509
510impl PartialEq for ObserverId {
511    fn eq(&self, other: &Self) -> bool {
512        Rc::ptr_eq(&self.0, &other.0)
513    }
514}
515
516impl Eq for ObserverId {}
517
518impl Hash for ObserverId {
519    fn hash<H: Hasher>(&self, state: &mut H) {
520        Rc::as_ptr(&self.0).hash(state);
521    }
522}
523
524thread_local! {
525    static APPLY_OBSERVERS: RefCell<HashMap<ObserverId, ApplyObserver>> = RefCell::new(HashMap::default());
526}
527
528thread_local! {
529    static LAST_WRITES: RefCell<HashMap<StateObjectId, SnapshotId>> = RefCell::new(HashMap::default());
530}
531
532thread_local! {
533    static EXTRA_STATE_OBJECTS: RefCell<crate::snapshot_weak_set::SnapshotWeakSet> = RefCell::new(crate::snapshot_weak_set::SnapshotWeakSet::new());
534}
535
536const UNUSED_RECORD_CLEANUP_INTERVAL: SnapshotId = 2;
537const UNUSED_RECORD_CLEANUP_BUSY_INTERVAL: SnapshotId = 1;
538const UNUSED_RECORD_CLEANUP_MIN_SIZE: usize = 64;
539
540thread_local! {
541    static LAST_UNUSED_RECORD_CLEANUP: Cell<SnapshotId> = const { Cell::new(0) };
542}
543
544#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
545pub struct SnapshotV2DebugStats {
546    pub apply_observers_len: usize,
547    pub apply_observers_cap: usize,
548    pub last_writes_len: usize,
549    pub last_writes_cap: usize,
550    pub extra_state_objects_len: usize,
551    pub extra_state_objects_cap: usize,
552    pub last_unused_record_cleanup: SnapshotId,
553}
554
555pub fn debug_snapshot_v2_stats() -> SnapshotV2DebugStats {
556    let (apply_observers_len, apply_observers_cap) = APPLY_OBSERVERS.with(|cell| {
557        let observers = cell.borrow();
558        (observers.len(), observers.capacity())
559    });
560    let (last_writes_len, last_writes_cap) = LAST_WRITES.with(|cell| {
561        let writes = cell.borrow();
562        (writes.len(), writes.capacity())
563    });
564    let SnapshotWeakSetDebugStats {
565        len: extra_state_objects_len,
566        capacity: extra_state_objects_cap,
567    } = EXTRA_STATE_OBJECTS.with(|cell| cell.borrow().debug_stats());
568    let last_unused_record_cleanup = LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.get());
569
570    SnapshotV2DebugStats {
571        apply_observers_len,
572        apply_observers_cap,
573        last_writes_len,
574        last_writes_cap,
575        extra_state_objects_len,
576        extra_state_objects_cap,
577        last_unused_record_cleanup,
578    }
579}
580
581/// Register an apply observer.
582///
583/// Returns a handle that will automatically unregister the observer when dropped.
584pub fn register_apply_observer(observer: ApplyObserver) -> ObserverHandle {
585    let id = ObserverId::new();
586    APPLY_OBSERVERS.with(|cell| {
587        cell.borrow_mut().insert(id.clone(), observer);
588    });
589    ObserverHandle {
590        kind: ObserverKind::Apply,
591        id,
592    }
593}
594
595/// Handle for unregistering observers.
596///
597/// When dropped, automatically removes the associated observer.
598pub struct ObserverHandle {
599    kind: ObserverKind,
600    id: ObserverId,
601}
602
603enum ObserverKind {
604    Apply,
605}
606
607impl Drop for ObserverHandle {
608    fn drop(&mut self) {
609        match self.kind {
610            ObserverKind::Apply => {
611                APPLY_OBSERVERS.with(|cell| {
612                    cell.borrow_mut().remove(&self.id);
613                });
614            }
615        }
616    }
617}
618
619pub(crate) fn notify_apply_observers(modified: &[Arc<dyn StateObject>], snapshot_id: SnapshotId) {
620    APPLY_OBSERVERS.with(|cell| {
621        let observers: Vec<ApplyObserver> = cell.borrow().values().cloned().collect();
622        for observer in observers.into_iter() {
623            observer(modified, snapshot_id);
624        }
625    });
626}
627
628pub(crate) fn set_last_write(id: StateObjectId, snapshot_id: SnapshotId) {
629    LAST_WRITES.with(|cell| {
630        cell.borrow_mut().insert(id, snapshot_id);
631    });
632}
633
634#[cfg(test)]
635pub(crate) fn clear_last_writes() {
636    LAST_WRITES.with(|cell| {
637        cell.borrow_mut().clear();
638    });
639}
640
641pub(crate) fn check_and_overwrite_unused_records_locked() {
642    EXTRA_STATE_OBJECTS.with(|cell| {
643        cell.borrow_mut()
644            .remove_if(|state| state.overwrite_unused_records());
645    });
646}
647
648pub(crate) fn maybe_check_and_overwrite_unused_records_locked(current_snapshot_id: SnapshotId) {
649    let should_run = EXTRA_STATE_OBJECTS.with(|cell| {
650        let set = cell.borrow();
651        if set.is_empty() {
652            return false;
653        }
654        let last_cleanup = LAST_UNUSED_RECORD_CLEANUP.with(|last| last.get());
655        let interval = if set.len() >= UNUSED_RECORD_CLEANUP_MIN_SIZE {
656            UNUSED_RECORD_CLEANUP_BUSY_INTERVAL
657        } else {
658            UNUSED_RECORD_CLEANUP_INTERVAL
659        };
660        current_snapshot_id.saturating_sub(last_cleanup) >= interval
661    });
662
663    if should_run {
664        LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.set(current_snapshot_id));
665        check_and_overwrite_unused_records_locked();
666    }
667}
668
669#[cfg(test)]
670pub(crate) fn clear_unused_record_cleanup_for_tests() {
671    LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.set(0));
672}
673
674pub(crate) fn optimistic_merges(
675    current_snapshot_id: SnapshotId,
676    base_parent_id: SnapshotId,
677    modified_objects: &[(StateObjectId, Arc<dyn StateObject>, SnapshotId)],
678    invalid_snapshots: &SnapshotIdSet,
679    applying_invalid: &SnapshotIdSet,
680) -> Option<HashMap<usize, Rc<StateRecord>>> {
681    if modified_objects.is_empty() {
682        return None;
683    }
684
685    let mut result: Option<HashMap<usize, Rc<StateRecord>>> = None;
686
687    for (_, state, writer_id) in modified_objects.iter() {
688        let head = state.first_record();
689
690        let current = match crate::state::readable_record_for(
691            &head,
692            current_snapshot_id,
693            invalid_snapshots,
694        ) {
695            Some(record) => record,
696            None => continue,
697        };
698
699        let (previous_opt, found_base) =
700            mutable::find_previous_record(&head, base_parent_id, applying_invalid);
701        let previous = previous_opt?;
702
703        if !found_base || previous.snapshot_id() == crate::state::PREEXISTING_SNAPSHOT_ID {
704            continue;
705        }
706
707        if Rc::ptr_eq(&current, &previous) {
708            continue;
709        }
710
711        let applied = mutable::find_record_by_id(&head, *writer_id)?;
712
713        let merged = state.merge_records(
714            Rc::clone(&previous),
715            Rc::clone(&current),
716            Rc::clone(&applied),
717        )?;
718
719        result
720            .get_or_insert_with(HashMap::default)
721            .insert(Rc::as_ptr(&current) as usize, merged);
722    }
723
724    result
725}
726
727#[allow(clippy::arc_with_non_send_sync)]
728fn merge_observers(a: Option<ReadObserver>, b: Option<ReadObserver>) -> Option<ReadObserver> {
729    match (a, b) {
730        (None, None) => None,
731        (Some(a), None) => Some(a),
732        (None, Some(b)) => Some(b),
733        (Some(a), Some(b)) => Some(Arc::new(move |state: &dyn StateObject| {
734            a(state);
735            b(state);
736        })),
737    }
738}
739
740/// Merge two read observers into one.
741///
742/// # Thread Safety
743/// The resulting Arc-wrapped closure may capture non-Send closures. This is safe
744/// because observers are only invoked on the UI thread where they were created.
745pub fn merge_read_observers(
746    a: Option<ReadObserver>,
747    b: Option<ReadObserver>,
748) -> Option<ReadObserver> {
749    merge_observers(a, b)
750}
751
752/// Merge two write observers into one.
753///
754/// # Thread Safety
755/// The resulting Arc-wrapped closure may capture non-Send closures. This is safe
756/// because observers are only invoked on the UI thread where they were created.
757pub fn merge_write_observers(
758    a: Option<WriteObserver>,
759    b: Option<WriteObserver>,
760) -> Option<WriteObserver> {
761    merge_observers(a, b)
762}
763
764pub(crate) struct SnapshotState {
765    pub(crate) id: Cell<SnapshotId>,
766    pub(crate) invalid: RefCell<SnapshotIdSet>,
767    pub(crate) pin_handle: Cell<PinHandle>,
768    pub(crate) disposed: Cell<bool>,
769    pub(crate) read_observer: RefCell<Option<ReadObserver>>,
770    pub(crate) write_observer: RefCell<Option<WriteObserver>>,
771    #[allow(clippy::type_complexity)]
772    pub(crate) modified: RefCell<HashMap<StateObjectId, (Arc<dyn StateObject>, SnapshotId)>>,
773    on_dispose: RefCell<Option<Box<dyn FnOnce()>>>,
774    runtime_tracked: bool,
775    pending_children: RefCell<HashSet<SnapshotId>>,
776}
777
778impl SnapshotState {
779    pub(crate) fn new(
780        id: SnapshotId,
781        invalid: SnapshotIdSet,
782        read_observer: Option<ReadObserver>,
783        write_observer: Option<WriteObserver>,
784        runtime_tracked: bool,
785    ) -> Self {
786        Self::new_with_pinning(
787            id,
788            invalid,
789            read_observer,
790            write_observer,
791            runtime_tracked,
792            true,
793        )
794    }
795
796    pub(crate) fn new_with_pinning(
797        id: SnapshotId,
798        invalid: SnapshotIdSet,
799        read_observer: Option<ReadObserver>,
800        write_observer: Option<WriteObserver>,
801        runtime_tracked: bool,
802        should_pin: bool,
803    ) -> Self {
804        let pin_handle = if should_pin {
805            snapshot_pinning::track_pinning(id, &invalid)
806        } else {
807            snapshot_pinning::PinHandle::INVALID
808        };
809        Self {
810            id: Cell::new(id),
811            invalid: RefCell::new(invalid),
812            pin_handle: Cell::new(pin_handle),
813            disposed: Cell::new(false),
814            read_observer: RefCell::new(read_observer),
815            write_observer: RefCell::new(write_observer),
816            modified: RefCell::new(HashMap::default()),
817            on_dispose: RefCell::new(None),
818            runtime_tracked,
819            pending_children: RefCell::new(HashSet::default()),
820        }
821    }
822
823    pub(crate) fn record_read(&self, state: &dyn StateObject) {
824        if let Some(observer) = self.read_observer.borrow().as_ref() {
825            observer(state);
826        }
827    }
828
829    pub(crate) fn record_write(&self, state: Arc<dyn StateObject>, writer_id: SnapshotId) {
830        let state_id = state.object_id().as_usize();
831
832        let mut modified = self.modified.borrow_mut();
833
834        match modified.entry(state_id) {
835            std::collections::hash_map::Entry::Vacant(e) => {
836                if let Some(observer) = self.write_observer.borrow().as_ref() {
837                    observer(&*state);
838                }
839                e.insert((state, writer_id));
840            }
841            std::collections::hash_map::Entry::Occupied(mut e) => {
842                e.insert((state, writer_id));
843            }
844        }
845    }
846
847    pub(crate) fn dispose(&self) {
848        if !self.disposed.replace(true) {
849            let pin_handle = self.pin_handle.get();
850            snapshot_pinning::release_pinning(pin_handle);
851            if let Some(cb) = self.on_dispose.borrow_mut().take() {
852                cb();
853            }
854            if self.runtime_tracked {
855                close_snapshot(self.id.get());
856            }
857        }
858    }
859
860    pub(crate) fn add_pending_child(&self, id: SnapshotId) {
861        self.pending_children.borrow_mut().insert(id);
862    }
863
864    pub(crate) fn remove_pending_child(&self, id: SnapshotId) {
865        self.pending_children.borrow_mut().remove(&id);
866    }
867
868    pub(crate) fn has_pending_children(&self) -> bool {
869        !self.pending_children.borrow().is_empty()
870    }
871
872    pub(crate) fn pending_children(&self) -> Vec<SnapshotId> {
873        self.pending_children.borrow().iter().copied().collect()
874    }
875
876    pub(crate) fn set_on_dispose<F>(&self, f: F)
877    where
878        F: FnOnce() + 'static,
879    {
880        *self.on_dispose.borrow_mut() = Some(Box::new(f));
881    }
882}
883
884pub(crate) trait NestedMutableHost {
885    fn snapshot_state(&self) -> &SnapshotState;
886    fn nested_count(&self) -> &Cell<usize>;
887}
888
889pub(crate) fn clear_nested_child_on_dispose<P>(
890    parent: &Arc<P>,
891    child_id: SnapshotId,
892) -> impl FnOnce() + 'static
893where
894    P: NestedMutableHost + 'static,
895{
896    let weak = Arc::downgrade(parent);
897    move || {
898        if let Some(parent) = weak.upgrade() {
899            let nested_count = parent.nested_count();
900            if nested_count.get() > 0 {
901                nested_count.set(nested_count.get().saturating_sub(1));
902            }
903            let state = parent.snapshot_state();
904            let new_invalid = state.invalid.borrow().clone().clear(child_id);
905            state.invalid.replace(new_invalid);
906            state.remove_pending_child(child_id);
907        }
908    }
909}
910
911pub(crate) fn allocate_nested_mutable_snapshot<P>(
912    parent: &Arc<P>,
913    root: Weak<MutableSnapshot>,
914    read_observer: Option<ReadObserver>,
915    write_observer: Option<WriteObserver>,
916) -> Arc<NestedMutableSnapshot>
917where
918    P: NestedMutableHost + 'static,
919{
920    let state = parent.snapshot_state();
921    let merged_read = merge_read_observers(read_observer, state.read_observer.borrow().clone());
922    let merged_write = merge_write_observers(write_observer, state.write_observer.borrow().clone());
923
924    let parent_id = state.id.get();
925    let current_invalid = state.invalid.borrow().clone();
926
927    let (new_id, _runtime_invalid) = allocate_snapshot();
928
929    let parent_invalid_with_child = current_invalid.set(new_id);
930    state.invalid.replace(parent_invalid_with_child);
931
932    let invalid = current_invalid.add_range(parent_id + 1, new_id);
933
934    let nested = NestedMutableSnapshot::new(
935        new_id,
936        invalid,
937        merged_read,
938        merged_write,
939        root,
940        state.id.get(),
941    );
942
943    let nested_count = parent.nested_count();
944    nested_count.set(nested_count.get() + 1);
945    state.add_pending_child(new_id);
946
947    nested.set_on_dispose(clear_nested_child_on_dispose(parent, new_id));
948
949    nested
950}
951
952#[cfg(test)]
953mod tests {
954    use std::sync::Mutex;
955
956    use super::*;
957
958    #[test]
959    fn apply_observer_ids_do_not_use_process_global_counter() {
960        let source = include_str!("mod.rs");
961        assert!(!source.contains(concat!("NEXT_", "OBSERVER_ID")));
962        assert!(!source.contains(concat!("Atomic", "Usize")));
963    }
964
965    #[test]
966    fn test_apply_result_is_success() {
967        assert!(SnapshotApplyResult::Success.is_success());
968        assert!(!SnapshotApplyResult::Failure.is_success());
969    }
970
971    #[test]
972    fn test_apply_result_is_failure() {
973        assert!(!SnapshotApplyResult::Success.is_failure());
974        assert!(SnapshotApplyResult::Failure.is_failure());
975    }
976
977    #[test]
978    fn test_apply_result_check_success() {
979        SnapshotApplyResult::Success.check();
980    }
981
982    #[test]
983    #[should_panic(expected = "Snapshot apply failed")]
984    fn test_apply_result_check_failure() {
985        SnapshotApplyResult::Failure.check();
986    }
987
988    #[test]
989    fn snapshot_enter_restores_current_snapshot_after_panic() {
990        let _guard = reset_runtime_for_tests();
991        let snapshot = take_mutable_snapshot(None, None);
992
993        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
994            snapshot.enter(|| panic!("snapshot body panic"));
995        }));
996
997        assert!(result.is_err());
998        assert!(
999            current_snapshot().is_none(),
1000            "snapshot enter must restore the previous current snapshot while unwinding"
1001        );
1002    }
1003
1004    #[test]
1005    fn test_merge_read_observers_both_none() {
1006        let result = merge_read_observers(None, None);
1007        assert!(result.is_none());
1008    }
1009
1010    #[test]
1011    fn test_merge_read_observers_one_some() {
1012        let observer = Arc::new(|_: &dyn StateObject| {});
1013        let result = merge_read_observers(Some(observer.clone()), None);
1014        assert!(result.is_some());
1015
1016        let result = merge_read_observers(None, Some(observer));
1017        assert!(result.is_some());
1018    }
1019
1020    #[test]
1021    fn test_merge_write_observers_both_none() {
1022        let result = merge_write_observers(None, None);
1023        assert!(result.is_none());
1024    }
1025
1026    #[test]
1027    fn test_merge_write_observers_one_some() {
1028        let observer = Arc::new(|_: &dyn StateObject| {});
1029        let result = merge_write_observers(Some(observer.clone()), None);
1030        assert!(result.is_some());
1031
1032        let result = merge_write_observers(None, Some(observer));
1033        assert!(result.is_some());
1034    }
1035
1036    #[test]
1037    fn test_current_snapshot_none_initially() {
1038        set_current_snapshot(None);
1039        assert!(current_snapshot().is_none());
1040    }
1041
1042    struct TestStateObject {
1043        id: usize,
1044    }
1045
1046    impl TestStateObject {
1047        fn new(id: usize) -> Arc<Self> {
1048            Arc::new(Self { id })
1049        }
1050    }
1051
1052    impl StateObject for TestStateObject {
1053        fn object_id(&self) -> crate::state::ObjectId {
1054            crate::state::ObjectId(self.id)
1055        }
1056
1057        fn first_record(&self) -> Rc<crate::state::StateRecord> {
1058            unimplemented!("Not needed for observer tests")
1059        }
1060
1061        fn try_readable_record(
1062            &self,
1063            _snapshot_id: SnapshotId,
1064            _invalid: &SnapshotIdSet,
1065        ) -> Option<Rc<crate::state::StateRecord>> {
1066            None
1067        }
1068
1069        fn readable_record(
1070            &self,
1071            _snapshot_id: SnapshotId,
1072            _invalid: &SnapshotIdSet,
1073        ) -> Rc<crate::state::StateRecord> {
1074            unimplemented!("Not needed for observer tests")
1075        }
1076
1077        fn prepend_state_record(&self, _record: Rc<crate::state::StateRecord>) {
1078            unimplemented!("Not needed for observer tests")
1079        }
1080
1081        fn promote_record(&self, _child_id: SnapshotId) -> Result<(), &'static str> {
1082            unimplemented!("Not needed for observer tests")
1083        }
1084
1085        fn as_any(&self) -> &dyn std::any::Any {
1086            self
1087        }
1088    }
1089
1090    #[test]
1091    fn test_apply_observer_receives_correct_modified_objects() {
1092        use std::sync::Mutex;
1093
1094        let received_count = Arc::new(Mutex::new(0));
1095        let received_snapshot_id = Arc::new(Mutex::new(0));
1096
1097        let received_count_clone = received_count.clone();
1098        let received_snapshot_id_clone = received_snapshot_id.clone();
1099
1100        let _handle = register_apply_observer(Rc::new(move |modified, snapshot_id| {
1101            *received_snapshot_id_clone.lock().unwrap() = snapshot_id;
1102            *received_count_clone.lock().unwrap() = modified.len();
1103        }));
1104
1105        let obj1: Arc<dyn StateObject> = TestStateObject::new(42);
1106        let obj2: Arc<dyn StateObject> = TestStateObject::new(99);
1107        let modified = vec![obj1, obj2];
1108
1109        notify_apply_observers(&modified, 123);
1110
1111        assert_eq!(*received_snapshot_id.lock().unwrap(), 123);
1112        assert_eq!(*received_count.lock().unwrap(), 2);
1113    }
1114
1115    #[test]
1116    fn test_apply_observer_receives_correct_snapshot_id() {
1117        use std::sync::Mutex;
1118
1119        let received_id = Arc::new(Mutex::new(0));
1120        let received_id_clone = received_id.clone();
1121
1122        let _handle = register_apply_observer(Rc::new(move |_, snapshot_id| {
1123            *received_id_clone.lock().unwrap() = snapshot_id;
1124        }));
1125
1126        notify_apply_observers(&[], 456);
1127
1128        assert_eq!(*received_id.lock().unwrap(), 456);
1129    }
1130
1131    #[test]
1132    fn test_multiple_apply_observers_all_called() {
1133        use std::sync::Mutex;
1134
1135        let call_count1 = Arc::new(Mutex::new(0));
1136        let call_count2 = Arc::new(Mutex::new(0));
1137        let call_count3 = Arc::new(Mutex::new(0));
1138
1139        let call_count1_clone = call_count1.clone();
1140        let call_count2_clone = call_count2.clone();
1141        let call_count3_clone = call_count3.clone();
1142
1143        let _handle1 = register_apply_observer(Rc::new(move |_, _| {
1144            *call_count1_clone.lock().unwrap() += 1;
1145        }));
1146
1147        let _handle2 = register_apply_observer(Rc::new(move |_, _| {
1148            *call_count2_clone.lock().unwrap() += 1;
1149        }));
1150
1151        let _handle3 = register_apply_observer(Rc::new(move |_, _| {
1152            *call_count3_clone.lock().unwrap() += 1;
1153        }));
1154
1155        notify_apply_observers(&[], 1);
1156
1157        assert_eq!(*call_count1.lock().unwrap(), 1);
1158        assert_eq!(*call_count2.lock().unwrap(), 1);
1159        assert_eq!(*call_count3.lock().unwrap(), 1);
1160
1161        notify_apply_observers(&[], 2);
1162
1163        assert_eq!(*call_count1.lock().unwrap(), 2);
1164        assert_eq!(*call_count2.lock().unwrap(), 2);
1165        assert_eq!(*call_count3.lock().unwrap(), 2);
1166    }
1167
1168    #[test]
1169    fn test_apply_observer_not_called_for_empty_modifications() {
1170        use std::sync::Mutex;
1171
1172        let call_count = Arc::new(Mutex::new(0));
1173        let call_count_clone = call_count.clone();
1174
1175        let _handle = register_apply_observer(Rc::new(move |modified, _| {
1176            *call_count_clone.lock().unwrap() += 1;
1177            assert_eq!(modified.len(), 0);
1178        }));
1179
1180        notify_apply_observers(&[], 1);
1181
1182        assert_eq!(*call_count.lock().unwrap(), 1);
1183    }
1184
1185    fn register_counting_observer(calls: &Arc<Mutex<Vec<i32>>>, tag: i32) -> ObserverHandle {
1186        let calls = calls.clone();
1187        register_apply_observer(Rc::new(move |_, _| {
1188            calls.lock().unwrap().push(tag);
1189        }))
1190    }
1191
1192    #[test]
1193    fn test_observer_handle_drop_removes_correct_observer() {
1194        let calls = Arc::new(Mutex::new(Vec::new()));
1195
1196        let handle1 = register_counting_observer(&calls, 1);
1197        let handle2 = register_counting_observer(&calls, 2);
1198        let handle3 = register_counting_observer(&calls, 3);
1199
1200        notify_apply_observers(&[], 1);
1201        let result = calls.lock().unwrap().clone();
1202        assert_eq!(result.len(), 3);
1203        assert!(result.contains(&1));
1204        assert!(result.contains(&2));
1205        assert!(result.contains(&3));
1206        calls.lock().unwrap().clear();
1207
1208        drop(handle2);
1209
1210        notify_apply_observers(&[], 2);
1211        let result = calls.lock().unwrap().clone();
1212        assert_eq!(result.len(), 2);
1213        assert!(result.contains(&1));
1214        assert!(result.contains(&3));
1215        assert!(!result.contains(&2));
1216        calls.lock().unwrap().clear();
1217
1218        drop(handle1);
1219
1220        notify_apply_observers(&[], 3);
1221        let result = calls.lock().unwrap().clone();
1222        assert_eq!(result.len(), 1);
1223        assert!(result.contains(&3));
1224        calls.lock().unwrap().clear();
1225
1226        drop(handle3);
1227
1228        notify_apply_observers(&[], 4);
1229        assert_eq!(calls.lock().unwrap().len(), 0);
1230    }
1231
1232    #[test]
1233    fn test_observer_handle_drop_in_different_orders() {
1234        {
1235            let calls = Arc::new(Mutex::new(Vec::new()));
1236
1237            let h1 = register_counting_observer(&calls, 1);
1238            let h2 = register_counting_observer(&calls, 2);
1239            let h3 = register_counting_observer(&calls, 3);
1240
1241            drop(h3);
1242            notify_apply_observers(&[], 1);
1243            let result = calls.lock().unwrap().clone();
1244            assert!(result.contains(&1) && result.contains(&2) && !result.contains(&3));
1245            calls.lock().unwrap().clear();
1246
1247            drop(h2);
1248            notify_apply_observers(&[], 2);
1249            let result = calls.lock().unwrap().clone();
1250            assert_eq!(result.len(), 1);
1251            assert!(result.contains(&1));
1252            calls.lock().unwrap().clear();
1253
1254            drop(h1);
1255            notify_apply_observers(&[], 3);
1256            assert_eq!(calls.lock().unwrap().len(), 0);
1257        }
1258
1259        {
1260            let calls = Arc::new(Mutex::new(Vec::new()));
1261
1262            let h1 = register_counting_observer(&calls, 1);
1263            let h2 = register_counting_observer(&calls, 2);
1264            let h3 = register_counting_observer(&calls, 3);
1265
1266            drop(h1);
1267            notify_apply_observers(&[], 1);
1268            let result = calls.lock().unwrap().clone();
1269            assert!(!result.contains(&1) && result.contains(&2) && result.contains(&3));
1270            calls.lock().unwrap().clear();
1271
1272            drop(h2);
1273            notify_apply_observers(&[], 2);
1274            let result = calls.lock().unwrap().clone();
1275            assert_eq!(result.len(), 1);
1276            assert!(result.contains(&3));
1277            calls.lock().unwrap().clear();
1278
1279            drop(h3);
1280            notify_apply_observers(&[], 3);
1281            assert_eq!(calls.lock().unwrap().len(), 0);
1282        }
1283    }
1284
1285    #[test]
1286    fn test_remaining_observers_still_work_after_drop() {
1287        use std::sync::Mutex;
1288
1289        let calls = Arc::new(Mutex::new(Vec::new()));
1290
1291        let calls1 = calls.clone();
1292        let handle1 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1293            calls1.lock().unwrap().push((1, snapshot_id));
1294        }));
1295
1296        let calls2 = calls.clone();
1297        let handle2 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1298            calls2.lock().unwrap().push((2, snapshot_id));
1299        }));
1300
1301        notify_apply_observers(&[], 100);
1302        assert_eq!(calls.lock().unwrap().len(), 2);
1303        calls.lock().unwrap().clear();
1304
1305        drop(handle1);
1306
1307        notify_apply_observers(&[], 200);
1308        assert_eq!(*calls.lock().unwrap(), vec![(2, 200)]);
1309        calls.lock().unwrap().clear();
1310
1311        let calls3 = calls.clone();
1312        let _handle3 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1313            calls3.lock().unwrap().push((3, snapshot_id));
1314        }));
1315
1316        notify_apply_observers(&[], 300);
1317        let result = calls.lock().unwrap().clone();
1318        assert_eq!(result.len(), 2);
1319        assert!(result.contains(&(2, 300)));
1320        assert!(result.contains(&(3, 300)));
1321
1322        drop(handle2);
1323    }
1324
1325    #[test]
1326    fn test_observer_ids_are_unique() {
1327        use std::sync::Mutex;
1328
1329        let ids = Arc::new(Mutex::new(std::collections::HashSet::new()));
1330
1331        let mut handles = Vec::new();
1332
1333        for i in 0..100 {
1334            let ids_clone = ids.clone();
1335            let handle = register_apply_observer(Rc::new(move |_, _| {
1336                ids_clone.lock().unwrap().insert(i);
1337            }));
1338            handles.push(handle);
1339        }
1340
1341        notify_apply_observers(&[], 1);
1342        assert_eq!(ids.lock().unwrap().len(), 100);
1343
1344        for i in (0..100).step_by(2) {
1345            handles.remove(i / 2);
1346        }
1347
1348        ids.lock().unwrap().clear();
1349        notify_apply_observers(&[], 2);
1350        assert_eq!(ids.lock().unwrap().len(), 50);
1351    }
1352
1353    #[test]
1354    fn test_state_object_storage_in_modified_set() {
1355        let state = SnapshotState::new(1, SnapshotIdSet::new(), None, None, false);
1356
1357        let state_obj = TestStateObject::new(12345) as Arc<dyn StateObject>;
1358
1359        state.record_write(state_obj.clone(), 1);
1360
1361        let modified = state.modified.borrow();
1362        assert_eq!(modified.len(), 1);
1363        assert!(modified.contains_key(&12345));
1364
1365        let (stored, writer_id) = modified.get(&12345).unwrap();
1366        assert_eq!(stored.object_id().as_usize(), 12345);
1367        assert_eq!(*writer_id, 1);
1368    }
1369
1370    #[test]
1371    fn test_multiple_writes_to_same_state_object() {
1372        let state = SnapshotState::new(1, SnapshotIdSet::new(), None, None, false);
1373        let state_obj = TestStateObject::new(99999) as Arc<dyn StateObject>;
1374
1375        state.record_write(state_obj.clone(), 1);
1376        assert_eq!(state.modified.borrow().len(), 1);
1377
1378        state.record_write(state_obj.clone(), 2);
1379        let modified = state.modified.borrow();
1380        assert_eq!(modified.len(), 1);
1381        assert!(modified.contains_key(&99999));
1382        let (_, writer_id) = modified.get(&99999).unwrap();
1383        assert_eq!(*writer_id, 2);
1384    }
1385}