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