Skip to main content

cranpose_core/
state.rs

1#![allow(clippy::arc_with_non_send_sync)]
2
3use std::{
4    any::Any,
5    cell::{Cell, RefCell},
6    fmt,
7    hash::Hash,
8    marker::PhantomData,
9    ops::Deref,
10    rc::{Rc, Weak as RcWeak},
11    sync::{Arc, Mutex, MutexGuard, Weak},
12};
13
14use crate::{
15    RecomposeScope, RecomposeScopeInner, RuntimeHandle, ScopeId, StateId,
16    collections::map::{HashMap, HashSet},
17    debug_trace::debug_record_scope_invalidation,
18    runtime,
19    snapshot_id_set::{SnapshotId, SnapshotIdSet},
20    snapshot_pinning::lowest_pinned_snapshot,
21    snapshot_v2::{
22        AnySnapshot, GlobalSnapshot, advance_global_snapshot, allocate_record_id, current_snapshot,
23    },
24    with_current_composer_opt,
25};
26
27pub(crate) const PREEXISTING_SNAPSHOT_ID: SnapshotId = 1;
28
29const INVALID_SNAPSHOT_ID: SnapshotId = 0;
30
31const SNAPSHOT_ID_MAX: SnapshotId = usize::MAX;
32
33#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug, Default)]
34pub struct ObjectId(pub(crate) usize);
35
36impl ObjectId {
37    pub(crate) fn new<T: ?Sized + 'static>(object: &Arc<T>) -> Self {
38        Self(Arc::as_ptr(object) as *const () as usize)
39    }
40
41    #[inline]
42    pub(crate) fn as_usize(self) -> usize {
43        self.0
44    }
45}
46
47pub struct StateRecord {
48    snapshot_id: Cell<SnapshotId>,
49    tombstone: Cell<bool>,
50    next: Cell<Option<Rc<StateRecord>>>,
51    value: RefCell<Option<Box<dyn Any>>>,
52}
53
54#[derive(Debug)]
55struct StateReadFailure {
56    state_id: ObjectId,
57    snapshot_id: SnapshotId,
58    fresh_snapshot_id: SnapshotId,
59    fresh_invalid: SnapshotIdSet,
60    record_chain: Vec<(SnapshotId, bool)>,
61}
62
63impl std::fmt::Display for StateReadFailure {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        write!(
66            f,
67            "Reading a state that was created after the snapshot was taken or in a snapshot that has not yet been applied\n\
68             state={:?}, snapshot_id={}, fresh_snapshot_id={}, fresh_invalid={:?}\n\
69             record_chain={:?}",
70            self.state_id,
71            self.snapshot_id,
72            self.fresh_snapshot_id,
73            self.fresh_invalid,
74            self.record_chain
75        )
76    }
77}
78
79#[derive(Debug, Clone, Copy, Eq, PartialEq)]
80pub(crate) enum StateRecordValueError {
81    MissingOrWrongType { expected: &'static str },
82}
83
84impl StateRecord {
85    pub(crate) fn new<T: Any>(
86        snapshot_id: SnapshotId,
87        value: T,
88        next: Option<Rc<StateRecord>>,
89    ) -> Rc<Self> {
90        Rc::new(Self {
91            snapshot_id: Cell::new(snapshot_id),
92            tombstone: Cell::new(false),
93            next: Cell::new(next),
94            value: RefCell::new(Some(Box::new(value))),
95        })
96    }
97
98    #[inline]
99    pub(crate) fn snapshot_id(&self) -> SnapshotId {
100        self.snapshot_id.get()
101    }
102
103    #[inline]
104    pub(crate) fn set_snapshot_id(&self, id: SnapshotId) {
105        self.snapshot_id.set(id);
106    }
107
108    #[inline]
109    pub(crate) fn next(&self) -> Option<Rc<StateRecord>> {
110        self.next.take().inspect(|record| {
111            self.next.set(Some(Rc::clone(record)));
112        })
113    }
114
115    #[inline]
116    pub(crate) fn set_next(&self, next: Option<Rc<StateRecord>>) {
117        self.next.set(next);
118    }
119
120    #[inline]
121    pub(crate) fn is_tombstone(&self) -> bool {
122        self.tombstone.get()
123    }
124
125    #[inline]
126    pub(crate) fn set_tombstone(&self, tombstone: bool) {
127        self.tombstone.set(tombstone);
128    }
129
130    pub(crate) fn clear_value(&self) {
131        self.value.borrow_mut().take();
132    }
133
134    pub(crate) fn replace_value<T: Any>(&self, new_value: T) {
135        *self.value.borrow_mut() = Some(Box::new(new_value));
136    }
137
138    pub(crate) fn with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> R {
139        self.try_with_value(f)
140            .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"))
141    }
142
143    pub(crate) fn try_with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
144        let guard = self.value.borrow();
145        let value = guard.as_ref().and_then(|boxed| boxed.downcast_ref::<T>())?;
146        Some(f(value))
147    }
148
149    #[cfg(test)]
150    pub(crate) fn clear_for_reuse(&self) {
151        self.clear_value();
152    }
153
154    pub(crate) fn assign_value<T: Any + Clone>(
155        &self,
156        source: &StateRecord,
157    ) -> Result<(), StateRecordValueError> {
158        let cloned_value = source.try_with_value(|value: &T| value.clone()).ok_or(
159            StateRecordValueError::MissingOrWrongType {
160                expected: std::any::type_name::<T>(),
161            },
162        )?;
163        self.replace_value(cloned_value);
164        Ok(())
165    }
166}
167
168impl Drop for StateRecord {
169    fn drop(&mut self) {
170        let mut next = self.next.take();
171        while let Some(node) = next {
172            match Rc::try_unwrap(node) {
173                Ok(record) => {
174                    next = record.next.take();
175                }
176                Err(_) => {
177                    break;
178                }
179            }
180        }
181    }
182}
183
184struct CurrentRecord {
185    head: RefCell<Rc<StateRecord>>,
186}
187
188impl CurrentRecord {
189    fn new(head: Rc<StateRecord>) -> Self {
190        Self {
191            head: RefCell::new(head),
192        }
193    }
194
195    fn clone_head(&self) -> Rc<StateRecord> {
196        self.head.borrow().clone()
197    }
198
199    fn replace(&self, new_head: Rc<StateRecord>) {
200        *self.head.borrow_mut() = new_head;
201    }
202
203    fn prepend(&self, record: Rc<StateRecord>) {
204        let current_head = self.clone_head();
205        record.set_next(Some(current_head));
206        self.replace(record);
207    }
208}
209
210#[inline]
211fn record_is_valid_for(
212    record: &Rc<StateRecord>,
213    snapshot_id: SnapshotId,
214    invalid: &SnapshotIdSet,
215) -> bool {
216    if record.is_tombstone() {
217        return false;
218    }
219
220    let candidate = record.snapshot_id();
221    if candidate == INVALID_SNAPSHOT_ID || candidate > snapshot_id {
222        return false;
223    }
224
225    candidate == snapshot_id || !invalid.get(candidate)
226}
227
228pub(crate) fn readable_record_for(
229    head: &Rc<StateRecord>,
230    snapshot_id: SnapshotId,
231    invalid: &SnapshotIdSet,
232) -> Option<Rc<StateRecord>> {
233    let mut best: Option<Rc<StateRecord>> = None;
234    let mut cursor = Some(Rc::clone(head));
235
236    while let Some(record) = cursor {
237        if record_is_valid_for(&record, snapshot_id, invalid) {
238            let replace = best
239                .as_ref()
240                .map(|current| current.snapshot_id() < record.snapshot_id())
241                .unwrap_or(true);
242            if replace {
243                best = Some(Rc::clone(&record));
244            }
245        }
246        cursor = record.next();
247    }
248
249    best
250}
251
252fn find_youngest_or<F>(head: &Rc<StateRecord>, predicate: F) -> Rc<StateRecord>
253where
254    F: Fn(&Rc<StateRecord>) -> bool,
255{
256    let mut current = Some(Rc::clone(head));
257    let mut youngest = Rc::clone(head);
258
259    while let Some(record) = current {
260        if predicate(&record) {
261            return record;
262        }
263        if youngest.snapshot_id() < record.snapshot_id() {
264            youngest = Rc::clone(&record);
265        }
266        current = record.next();
267    }
268
269    youngest
270}
271
272pub(crate) fn used_locked(head: &Rc<StateRecord>) -> Option<Rc<StateRecord>> {
273    let mut current = Some(Rc::clone(head));
274    let mut valid_record: Option<Rc<StateRecord>> = None;
275
276    let reuse_limit = lowest_pinned_snapshot()
277        .map(|lowest| lowest.saturating_sub(1))
278        .unwrap_or_else(|| allocate_record_id().saturating_sub(1));
279
280    let invalid = SnapshotIdSet::EMPTY;
281
282    while let Some(record) = current {
283        let current_id = record.snapshot_id();
284
285        if current_id == PREEXISTING_SNAPSHOT_ID {
286            current = record.next();
287            continue;
288        }
289
290        if current_id == INVALID_SNAPSHOT_ID {
291            return Some(record);
292        }
293
294        if record.is_tombstone() && current_id < reuse_limit {
295            return Some(record);
296        }
297
298        if record_is_valid_for(&record, reuse_limit, &invalid) {
299            if let Some(ref existing) = valid_record {
300                return Some(if current_id < existing.snapshot_id() {
301                    record
302                } else {
303                    Rc::clone(existing)
304                });
305            } else {
306                valid_record = Some(record.clone());
307            }
308        }
309
310        current = record.next();
311    }
312
313    None
314}
315
316pub(crate) fn new_overwritable_record_locked(state: &dyn StateObject) -> Rc<StateRecord> {
317    let state_head = state.first_record();
318
319    if let Some(reusable) = used_locked(&state_head) {
320        reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
321        return reusable;
322    }
323
324    let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
325
326    state.prepend_state_record(Rc::clone(&new_record));
327
328    new_record
329}
330
331pub(crate) fn new_overwritable_record_as_head_locked(state: &dyn StateObject) -> Rc<StateRecord> {
332    let head = state.first_record();
333
334    if let Some(reusable) = used_locked(&head) {
335        reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
336
337        if !Rc::ptr_eq(&head, &reusable) {
338            let mut cursor = Some(Rc::clone(&head));
339            let mut unlinked = false;
340
341            while let Some(node) = cursor {
342                let next = node.next();
343                if let Some(next_record) = next {
344                    if Rc::ptr_eq(&next_record, &reusable) {
345                        node.set_next(reusable.next());
346                        unlinked = true;
347                        break;
348                    }
349                    cursor = Some(next_record);
350                } else {
351                    break;
352                }
353            }
354
355            if !unlinked {
356                debug_assert!(
357                    false,
358                    "new_overwritable_record_as_head_locked: reusable record not found in chain"
359                );
360                let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
361                state.prepend_state_record(Rc::clone(&new_record));
362                return new_record;
363            }
364
365            state.prepend_state_record(Rc::clone(&reusable));
366        }
367
368        return reusable;
369    }
370
371    let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
372    state.prepend_state_record(Rc::clone(&new_record));
373    new_record
374}
375
376pub(crate) fn overwrite_unused_records_locked<T: Any + Clone>(state: &dyn StateObject) -> bool {
377    let head = state.first_record();
378    let mut current = Some(Rc::clone(&head));
379    let mut overwrite_record: Option<Rc<StateRecord>> = None;
380    let mut valid_record: Option<Rc<StateRecord>> = None;
381
382    let reuse_limit =
383        lowest_pinned_snapshot().unwrap_or_else(crate::snapshot_v2::peek_next_snapshot_id);
384
385    let mut retained_records = 0;
386
387    while let Some(record) = current {
388        let current_id = record.snapshot_id();
389
390        if current_id == INVALID_SNAPSHOT_ID {
391        } else if current_id < reuse_limit {
392            if valid_record.is_none() {
393                valid_record = Some(Rc::clone(&record));
394                retained_records += 1;
395            } else {
396                let Some(valid) = valid_record.as_ref() else {
397                    valid_record = Some(Rc::clone(&record));
398                    retained_records += 1;
399                    current = record.next();
400                    continue;
401                };
402                let record_to_overwrite = if current_id < valid.snapshot_id() {
403                    Rc::clone(&record)
404                } else {
405                    let to_overwrite = Rc::clone(valid);
406                    valid_record = Some(Rc::clone(&record));
407                    to_overwrite
408                };
409
410                let source_record = overwrite_record.get_or_insert_with(|| {
411                    find_youngest_or(&head, |r| r.snapshot_id() >= reuse_limit)
412                });
413
414                record_to_overwrite.set_snapshot_id(INVALID_SNAPSHOT_ID);
415                if let Err(error) = record_to_overwrite.assign_value::<T>(source_record) {
416                    log::error!(
417                        "snapshot cleanup could not copy retained state record value for state {:?}: {:?}",
418                        state.object_id(),
419                        error
420                    );
421                }
422            }
423        } else {
424            retained_records += 1;
425        }
426
427        current = record.next();
428    }
429
430    retained_records > 1
431}
432
433fn active_snapshot() -> AnySnapshot {
434    current_snapshot().unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()))
435}
436
437pub(crate) trait MutationPolicy<T>: Send + Sync {
438    fn equivalent(&self, a: &T, b: &T) -> bool;
439    fn merge(&self, _previous: &T, _current: &T, _applied: &T) -> Option<T> {
440        None
441    }
442}
443
444pub(crate) struct NeverEqual;
445
446impl<T> MutationPolicy<T> for NeverEqual {
447    fn equivalent(&self, _a: &T, _b: &T) -> bool {
448        false
449    }
450}
451
452pub(crate) struct StructuralEqual;
453
454impl<T: PartialEq> MutationPolicy<T> for StructuralEqual {
455    fn equivalent(&self, a: &T, b: &T) -> bool {
456        a == b
457    }
458}
459
460pub trait StateObject: Any {
461    fn object_id(&self) -> ObjectId;
462    fn first_record(&self) -> Rc<StateRecord>;
463    fn try_readable_record(
464        &self,
465        snapshot_id: SnapshotId,
466        invalid: &SnapshotIdSet,
467    ) -> Option<Rc<StateRecord>>;
468    fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord>;
469
470    fn prepend_state_record(&self, record: Rc<StateRecord>);
471
472    fn merge_records(
473        &self,
474        _previous: Rc<StateRecord>,
475        _current: Rc<StateRecord>,
476        _applied: Rc<StateRecord>,
477    ) -> Option<Rc<StateRecord>> {
478        None
479    }
480
481    fn commit_merged_record(&self, _merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
482        Err("StateObject does not support merged record commits")
483    }
484    fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str>;
485
486    fn overwrite_unused_records(&self) -> bool {
487        false
488    }
489
490    fn observation_lease(&self) -> Option<Box<dyn Any>> {
491        None
492    }
493
494    fn as_any(&self) -> &dyn Any;
495}
496
497pub(crate) struct SnapshotMutableState<T> {
498    head: CurrentRecord,
499    policy: Arc<dyn MutationPolicy<T>>,
500    id: ObjectId,
501    weak_self: Mutex<Option<Weak<Self>>>,
502    apply_observers: Mutex<Vec<Box<dyn Fn() + 'static>>>,
503    read_observation_count: Cell<usize>,
504    scope_observation_count: Cell<usize>,
505    subscriber_callbacks: RefCell<Vec<RcWeak<dyn Fn()>>>,
506}
507
508struct StateObservationLease<T: Clone + 'static> {
509    state: Weak<SnapshotMutableState<T>>,
510}
511
512impl<T: Clone + 'static> Drop for StateObservationLease<T> {
513    fn drop(&mut self) {
514        if let Some(state) = self.state.upgrade() {
515            let count = state
516                .read_observation_count
517                .get()
518                .checked_sub(1)
519                .expect("state observation lease count underflow");
520            state.read_observation_count.set(count);
521        }
522    }
523}
524
525impl<T> SnapshotMutableState<T> {
526    fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
527        if !should_check_chain_integrity() {
528            return;
529        }
530        let head = self.head.clone_head();
531        let mut cursor = Some(head);
532        let mut seen: HashSet<usize> = HashSet::default();
533        let mut ids = Vec::new();
534
535        while let Some(record) = cursor {
536            let addr = Rc::as_ptr(&record) as usize;
537            assert!(
538                seen.insert(addr),
539                "SnapshotMutableState::{} detected duplicate/cycle at record {:p} for state {:?} (snapshot_context={:?}, chain_ids={:?})",
540                caller,
541                Rc::as_ptr(&record),
542                self.id,
543                snapshot_context,
544                ids
545            );
546            ids.push(record.snapshot_id());
547            cursor = record.next();
548        }
549
550        assert!(
551            !ids.is_empty(),
552            "SnapshotMutableState::{} finished integrity scan with empty id list for state {:?} (snapshot_context={:?})",
553            caller,
554            self.id,
555            snapshot_context
556        );
557    }
558}
559
560fn should_check_chain_integrity() -> bool {
561    #[cfg(debug_assertions)]
562    {
563        true
564    }
565
566    #[cfg(not(debug_assertions))]
567    {
568        crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
569    }
570}
571
572impl<T: Clone + 'static> SnapshotMutableState<T> {
573    fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
574        let mut chain_ids = Vec::new();
575        let mut cursor = Some(self.first_record());
576        while let Some(record) = cursor {
577            chain_ids.push((record.snapshot_id(), record.is_tombstone()));
578            cursor = record.next();
579        }
580        chain_ids
581    }
582
583    fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
584        let snapshot = active_snapshot();
585        if let Some(state) = self.upgrade_self() {
586            snapshot.record_read(&*state);
587        }
588
589        let snapshot_id = snapshot.snapshot_id();
590        let invalid = snapshot.invalid();
591
592        if let Some(record) = self.readable_for(snapshot_id, &invalid) {
593            return Ok(record);
594        }
595
596        let fresh_snapshot = active_snapshot();
597        let fresh_id = fresh_snapshot.snapshot_id();
598        let fresh_invalid = fresh_snapshot.invalid();
599
600        if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
601            return Ok(record);
602        }
603
604        let global = GlobalSnapshot::get_or_create();
605        let global_id = global.snapshot_id();
606        let global_invalid = global.invalid();
607
608        if let Some(record) = self.readable_for(global_id, &global_invalid) {
609            return Ok(record);
610        }
611
612        Err(StateReadFailure {
613            state_id: self.id,
614            snapshot_id,
615            fresh_snapshot_id: fresh_id,
616            fresh_invalid,
617            record_chain: self.record_chain_debug(),
618        })
619    }
620
621    fn readable_for(
622        &self,
623        snapshot_id: SnapshotId,
624        invalid: &SnapshotIdSet,
625    ) -> Option<Rc<StateRecord>> {
626        let head = self.first_record();
627        readable_record_for(&head, snapshot_id, invalid)
628    }
629
630    fn is_equivalent_to_readable(
631        &self,
632        snapshot_id: SnapshotId,
633        invalid: &SnapshotIdSet,
634        new_value: &T,
635    ) -> bool {
636        self.readable_for(snapshot_id, invalid)
637            .map(|record| {
638                record.with_value(|current: &T| self.policy.equivalent(current, new_value))
639            })
640            .unwrap_or(false)
641    }
642
643    fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
644        let readable = match self.readable_for(snapshot_id, invalid) {
645            Some(record) => record,
646            None => {
647                let current_head = self.head.clone_head();
648                let refreshed = readable_record_for(&current_head, snapshot_id, invalid);
649                let source = refreshed.unwrap_or_else(|| current_head.clone());
650
651                let cloned_value = source.with_value(|value: &T| value.clone());
652                let new_head = StateRecord::new(snapshot_id, cloned_value, Some(current_head));
653                self.head.replace(new_head.clone());
654                self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
655                return new_head;
656            }
657        };
658
659        if readable.snapshot_id() == snapshot_id {
660            return readable;
661        }
662
663        let refreshed = {
664            let current_head = self.head.clone_head();
665            let refreshed = readable_record_for(&current_head, snapshot_id, invalid).unwrap_or_else(
666                || {
667                    panic!(
668                        "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
669                        self.id, snapshot_id, invalid
670                    )
671                },
672            );
673
674            if refreshed.snapshot_id() == snapshot_id {
675                return refreshed;
676            }
677
678            Rc::clone(&refreshed)
679        };
680
681        let overwritable = new_overwritable_record_locked(self);
682        if let Err(error) = overwritable.assign_value::<T>(&refreshed) {
683            log::error!(
684                "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
685                self.id,
686                error
687            );
688        }
689        overwritable.set_snapshot_id(snapshot_id);
690        overwritable.set_tombstone(false);
691
692        self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
693
694        overwritable
695    }
696
697    pub(crate) fn new_in_arc(initial: T, policy: Arc<dyn MutationPolicy<T>>) -> Arc<Self> {
698        let snapshot = active_snapshot();
699        let snapshot_id = snapshot.snapshot_id();
700
701        let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
702        let head = StateRecord::new(snapshot_id, initial, Some(tail));
703
704        let mut state = Arc::new(Self {
705            head: CurrentRecord::new(head),
706            policy,
707            id: ObjectId::default(),
708            weak_self: Mutex::new(None),
709            apply_observers: Mutex::new(Vec::new()),
710            read_observation_count: Cell::new(0),
711            scope_observation_count: Cell::new(0),
712            subscriber_callbacks: RefCell::new(Vec::new()),
713        });
714
715        let id = ObjectId::new(&state);
716        if let Some(state_inner) = Arc::get_mut(&mut state) {
717            state_inner.id = id;
718        }
719
720        *state.lock_weak_self() = Some(Arc::downgrade(&state));
721
722        state
723    }
724
725    pub(crate) fn add_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
726        self.lock_apply_observers().push(observer);
727    }
728
729    fn acquire_observation_lease(&self) -> Option<Box<dyn Any>> {
730        let state = self.lock_weak_self().as_ref()?.clone();
731        let was_empty = !self.has_subscribers();
732        let count = self
733            .read_observation_count
734            .get()
735            .checked_add(1)
736            .expect("state observation lease count overflow");
737        self.read_observation_count.set(count);
738        if was_empty {
739            notify_subscriber_callbacks(&self.subscriber_callbacks);
740        }
741        Some(Box::new(StateObservationLease { state }))
742    }
743
744    fn add_scope_observer(&self) -> bool {
745        let was_empty = !self.has_subscribers();
746        let count = self
747            .scope_observation_count
748            .get()
749            .checked_add(1)
750            .expect("state scope observation count overflow");
751        self.scope_observation_count.set(count);
752        was_empty
753    }
754
755    fn remove_scope_observers(&self, count: usize) {
756        if count == 0 {
757            return;
758        }
759        let remaining = self
760            .scope_observation_count
761            .get()
762            .checked_sub(count)
763            .expect("state scope observation count underflow");
764        self.scope_observation_count.set(remaining);
765    }
766
767    fn has_subscribers(&self) -> bool {
768        self.read_observation_count.get() > 0 || self.scope_observation_count.get() > 0
769    }
770
771    fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
772        register_subscriber_callback(&self.subscriber_callbacks, &callback);
773        if notify {
774            callback();
775        }
776        drop(callback);
777        self.subscriber_callbacks
778            .borrow_mut()
779            .retain(|callback| callback.upgrade().is_some());
780    }
781
782    #[cfg(test)]
783    fn subscriber_callback_count(&self) -> usize {
784        self.subscriber_callbacks.borrow().len()
785    }
786
787    fn notify_subscribers(&self) {
788        notify_subscriber_callbacks(&self.subscriber_callbacks);
789    }
790
791    fn notify_applied(&self) {
792        let observers = self.lock_apply_observers();
793        for observer in observers.iter() {
794            observer();
795        }
796    }
797
798    fn lock_weak_self(&self) -> MutexGuard<'_, Option<Weak<Self>>> {
799        self.weak_self
800            .lock()
801            .unwrap_or_else(|poisoned| poisoned.into_inner())
802    }
803
804    fn lock_apply_observers(&self) -> MutexGuard<'_, Vec<Box<dyn Fn() + 'static>>> {
805        self.apply_observers
806            .lock()
807            .unwrap_or_else(|poisoned| poisoned.into_inner())
808    }
809
810    fn upgrade_self(&self) -> Option<Arc<Self>> {
811        self.lock_weak_self()
812            .as_ref()
813            .and_then(|weak| weak.upgrade())
814    }
815
816    #[inline]
817    pub(crate) fn id(&self) -> ObjectId {
818        self.id
819    }
820
821    pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
822        let record = self.readable_record_for_active_snapshot().ok()?;
823        record.try_with_value(f)
824    }
825
826    pub(crate) fn try_get(&self) -> Option<T> {
827        self.try_with_value(Clone::clone)
828    }
829
830    pub(crate) fn get(&self) -> T {
831        let record = self
832            .readable_record_for_active_snapshot()
833            .unwrap_or_else(|failure| panic!("{failure}"));
834        record.with_value(|value: &T| value.clone())
835    }
836
837    pub(crate) fn set(&self, new_value: T) -> bool {
838        #[cfg(debug_assertions)]
839        {
840            let in_handler = crate::in_event_handler();
841            let in_snapshot = crate::in_applied_snapshot();
842            if in_handler && !in_snapshot {
843                log::warn!(
844                    target: "cranpose::state",
845                    "State modified in event handler without run_in_mutable_snapshot; \
846                     this can make updates invisible to other contexts. Wrap the handler \
847                     in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
848                    self.id
849                );
850            }
851        }
852
853        let snapshot = active_snapshot();
854        let snapshot_id = snapshot.snapshot_id();
855
856        match &snapshot {
857            AnySnapshot::Global(global) => {
858                let invalid = snapshot.invalid();
859                if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
860                    return false;
861                }
862
863                if global.has_pending_children() {
864                    panic!(
865                        "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
866                        global.pending_children(),
867                        self.id,
868                        snapshot_id
869                    );
870                }
871
872                let mut written_state: Option<Arc<dyn StateObject>> = None;
873                if let Some(state) = self.upgrade_self() {
874                    let trait_object: Arc<dyn StateObject> = state.clone();
875                    snapshot.record_write(trait_object.clone());
876                    written_state = Some(trait_object);
877                }
878                mark_update_write(self.id);
879
880                let new_id = allocate_record_id();
881                let record = new_overwritable_record_as_head_locked(self);
882                record.replace_value(new_value);
883                record.set_snapshot_id(new_id);
884                record.set_tombstone(false);
885                advance_global_snapshot(new_id);
886                self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
887
888                if !global.has_pending_children() {
889                    let mut cursor = record.next();
890                    while let Some(node) = cursor {
891                        if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
892                            node.clear_value();
893                            node.set_tombstone(true);
894                        }
895                        cursor = node.next();
896                    }
897                    self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
898                }
899
900                if let Some(modified) = written_state.as_ref() {
901                    crate::snapshot_v2::notify_apply_observers(
902                        std::slice::from_ref(modified),
903                        new_id,
904                    );
905                }
906            }
907            AnySnapshot::Mutable(_)
908            | AnySnapshot::NestedMutable(_)
909            | AnySnapshot::TransparentMutable(_) => {
910                let invalid = snapshot.invalid();
911                if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
912                    return false;
913                }
914
915                if let Some(state) = self.upgrade_self() {
916                    let trait_object: Arc<dyn StateObject> = state.clone();
917                    snapshot.record_write(trait_object);
918                }
919                mark_update_write(self.id);
920
921                let record = self.writable_record(snapshot_id, &invalid);
922                record.replace_value(new_value);
923                self.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
924            }
925            AnySnapshot::Readonly(_)
926            | AnySnapshot::NestedReadonly(_)
927            | AnySnapshot::TransparentReadonly(_) => {
928                panic!("Cannot write to a read-only snapshot");
929            }
930        }
931
932        true
933    }
934}
935
936thread_local! {
937    static ACTIVE_UPDATES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
938    static PENDING_WRITES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
939}
940
941pub(crate) struct UpdateScope {
942    id: ObjectId,
943    finished: bool,
944}
945
946impl UpdateScope {
947    pub(crate) fn new(id: ObjectId) -> Self {
948        ACTIVE_UPDATES.with(|active| {
949            active.borrow_mut().insert(id);
950        });
951        PENDING_WRITES.with(|pending| {
952            pending.borrow_mut().remove(&id);
953        });
954        Self {
955            id,
956            finished: false,
957        }
958    }
959
960    pub(crate) fn finish(mut self) -> bool {
961        self.finished = true;
962        ACTIVE_UPDATES.with(|active| {
963            active.borrow_mut().remove(&self.id);
964        });
965        PENDING_WRITES.with(|pending| pending.borrow_mut().remove(&self.id))
966    }
967}
968
969impl Drop for UpdateScope {
970    fn drop(&mut self) {
971        if self.finished {
972            return;
973        }
974        ACTIVE_UPDATES.with(|active| {
975            active.borrow_mut().remove(&self.id);
976        });
977        PENDING_WRITES.with(|pending| {
978            pending.borrow_mut().remove(&self.id);
979        });
980    }
981}
982
983fn mark_update_write(id: ObjectId) {
984    ACTIVE_UPDATES.with(|active| {
985        if active.borrow().contains(&id) {
986            PENDING_WRITES.with(|pending| {
987                pending.borrow_mut().insert(id);
988            });
989        }
990    });
991}
992
993impl<T: Clone + 'static> SnapshotMutableState<T> {
994    fn try_readable_record(
995        &self,
996        snapshot_id: SnapshotId,
997        invalid: &SnapshotIdSet,
998    ) -> Option<Rc<StateRecord>> {
999        self.readable_for(snapshot_id, invalid)
1000    }
1001}
1002
1003impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
1004    fn object_id(&self) -> ObjectId {
1005        self.id
1006    }
1007
1008    fn first_record(&self) -> Rc<StateRecord> {
1009        self.head.clone_head()
1010    }
1011
1012    fn try_readable_record(
1013        &self,
1014        snapshot_id: SnapshotId,
1015        invalid: &SnapshotIdSet,
1016    ) -> Option<Rc<StateRecord>> {
1017        self.try_readable_record(snapshot_id, invalid)
1018    }
1019
1020    fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
1021        self.try_readable_record(snapshot_id, invalid)
1022            .unwrap_or_else(|| {
1023                panic!(
1024                    "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
1025                    self.id, snapshot_id
1026                )
1027            })
1028    }
1029
1030    fn prepend_state_record(&self, record: Rc<StateRecord>) {
1031        self.head.prepend(record);
1032    }
1033
1034    fn observation_lease(&self) -> Option<Box<dyn Any>> {
1035        self.acquire_observation_lease()
1036    }
1037
1038    fn merge_records(
1039        &self,
1040        previous: Rc<StateRecord>,
1041        current: Rc<StateRecord>,
1042        applied: Rc<StateRecord>,
1043    ) -> Option<Rc<StateRecord>> {
1044        let Some(current_value) = current.try_with_value(|value: &T| value.clone()) else {
1045            log::error!(
1046                "SnapshotMutableState::merge_records current record value missing or wrong type (state {:?}, current_id={})",
1047                self.id,
1048                current.snapshot_id()
1049            );
1050            return None;
1051        };
1052        let Some(applied_value) = applied.try_with_value(|value: &T| value.clone()) else {
1053            log::error!(
1054                "SnapshotMutableState::merge_records applied record value missing or wrong type (state {:?}, applied_id={})",
1055                self.id,
1056                applied.snapshot_id()
1057            );
1058            return None;
1059        };
1060        if self.policy.equivalent(&current_value, &applied_value) {
1061            return Some(current);
1062        }
1063
1064        let Some(previous_value) = previous.try_with_value(|value: &T| value.clone()) else {
1065            log::error!(
1066                "SnapshotMutableState::merge_records previous record value missing or wrong type (state {:?}, previous_id={})",
1067                self.id,
1068                previous.snapshot_id()
1069            );
1070            return None;
1071        };
1072        let merged = self
1073            .policy
1074            .merge(&previous_value, &current_value, &applied_value)?;
1075
1076        Some(StateRecord::new(applied.snapshot_id(), merged, None))
1077    }
1078
1079    fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1080        let head = self.first_record();
1081        let mut cursor = Some(head);
1082        while let Some(record) = cursor {
1083            if record.snapshot_id() == child_id {
1084                let Some(cloned) = record.try_with_value(|value: &T| value.clone()) else {
1085                    log::error!(
1086                        "SnapshotMutableState::promote_record child record value missing or wrong type (state {:?}, child_id={})",
1087                        self.id,
1088                        child_id
1089                    );
1090                    return Err("child record value missing or wrong type");
1091                };
1092                let new_id = allocate_record_id();
1093                let promoted = new_overwritable_record_as_head_locked(self);
1094                promoted.replace_value(cloned);
1095                promoted.set_tombstone(false);
1096                promoted.set_snapshot_id(new_id);
1097                advance_global_snapshot(new_id);
1098                self.notify_applied();
1099                self.assert_chain_integrity("promote_record", Some(child_id));
1100                return Ok(());
1101            }
1102            cursor = record.next();
1103        }
1104        log::error!(
1105            "SnapshotMutableState::promote_record missing child record (state {:?}, child_id={})",
1106            self.id,
1107            child_id
1108        );
1109        Err("missing child record")
1110    }
1111
1112    fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1113        let Some(value) = merged.try_with_value(|value: &T| value.clone()) else {
1114            log::error!(
1115                "SnapshotMutableState::commit_merged_record merged record value missing or wrong type (state {:?}, merged_id={})",
1116                self.id,
1117                merged.snapshot_id()
1118            );
1119            return Err("merged record value missing or wrong type");
1120        };
1121        let new_id = allocate_record_id();
1122        let committed = new_overwritable_record_as_head_locked(self);
1123        committed.replace_value(value);
1124        committed.set_tombstone(false);
1125        committed.set_snapshot_id(new_id);
1126        advance_global_snapshot(new_id);
1127        self.notify_applied();
1128        self.assert_chain_integrity("commit_merged_record", Some(new_id));
1129        Ok(new_id)
1130    }
1131
1132    fn overwrite_unused_records(&self) -> bool {
1133        overwrite_unused_records_locked::<T>(self)
1134    }
1135
1136    fn as_any(&self) -> &dyn Any {
1137        self
1138    }
1139}
1140
1141pub(crate) struct MutableStateInner<T: Clone + 'static> {
1142    pub(crate) state: Arc<SnapshotMutableState<T>>,
1143    pub(crate) watchers: RefCell<HashMap<ScopeId, RcWeak<RecomposeScopeInner>>>,
1144    runtime: RuntimeHandle,
1145    state_id: Cell<Option<StateId>>,
1146}
1147
1148fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<RcWeak<dyn Fn()>>>) {
1149    let callbacks_snapshot = std::mem::take(&mut *callbacks.borrow_mut());
1150    let mut live = Vec::with_capacity(callbacks_snapshot.len());
1151    for callback in callbacks_snapshot {
1152        let Some(callback) = callback.upgrade() else {
1153            continue;
1154        };
1155        callback();
1156        live.push(callback);
1157    }
1158    let mut registered = callbacks.borrow_mut();
1159    registered.retain(|callback| callback.upgrade().is_some());
1160    for callback in live {
1161        let callback = Rc::downgrade(&callback);
1162        if !registered
1163            .iter()
1164            .any(|registered| registered.ptr_eq(&callback))
1165        {
1166            registered.push(callback);
1167        }
1168    }
1169}
1170
1171fn register_subscriber_callback(
1172    callbacks: &RefCell<Vec<RcWeak<dyn Fn()>>>,
1173    callback: &Rc<dyn Fn()>,
1174) {
1175    let callback_weak = Rc::downgrade(callback);
1176    let mut callbacks = callbacks.borrow_mut();
1177    callbacks.retain(|callback| callback.upgrade().is_some());
1178    if !callbacks
1179        .iter()
1180        .any(|registered| registered.ptr_eq(&callback_weak))
1181    {
1182        callbacks.push(callback_weak);
1183    }
1184}
1185
1186fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, RcWeak<RecomposeScopeInner>>) {
1187    let len = watchers.len();
1188    let capacity = watchers.capacity();
1189    if capacity > len.saturating_mul(4).max(32) {
1190        watchers.shrink_to_fit();
1191    }
1192}
1193
1194impl<T: Clone + 'static> MutableStateInner<T> {
1195    pub(crate) fn new_with_policy(
1196        value: T,
1197        runtime: RuntimeHandle,
1198        policy: Arc<dyn MutationPolicy<T>>,
1199    ) -> Self {
1200        Self {
1201            state: SnapshotMutableState::new_in_arc(value, policy),
1202            watchers: RefCell::new(HashMap::default()),
1203            runtime,
1204            state_id: Cell::new(None),
1205        }
1206    }
1207
1208    pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1209        self.state_id.set(Some(state_id));
1210        let runtime_handle = self.runtime.clone();
1211        self.state.add_apply_observer(Box::new(move || {
1212            let runtime = runtime_handle.clone();
1213            runtime_handle.enqueue_ui_task(Box::new(move || {
1214                runtime.with_state_arena(|arena| {
1215                    let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1216                        inner.invalidate_watchers();
1217                    });
1218                });
1219            }));
1220        }));
1221    }
1222
1223    fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1224        let mut watchers = self.watchers.borrow_mut();
1225        let before = watchers.len();
1226        watchers.retain(|_, existing| existing.upgrade().is_some());
1227        self.state.remove_scope_observers(before - watchers.len());
1228        let registered = match watchers.get(&scope.id()) {
1229            Some(_) => false,
1230            _ => {
1231                watchers.insert(scope.id(), scope.downgrade());
1232                true
1233            }
1234        };
1235        drop(watchers);
1236        let became_subscribed = registered && self.state.add_scope_observer();
1237        (registered, became_subscribed)
1238    }
1239
1240    fn has_subscribers(&self) -> bool {
1241        let mut watchers = self.watchers.borrow_mut();
1242        let before = watchers.len();
1243        watchers.retain(|_, existing| existing.upgrade().is_some());
1244        self.state.remove_scope_observers(before - watchers.len());
1245        self.state.has_subscribers()
1246    }
1247
1248    pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1249        let mut watchers = self.watchers.borrow_mut();
1250        let removed = if watchers
1251            .get(&scope_id)
1252            .is_some_and(|weak| weak.upgrade().is_none())
1253        {
1254            watchers.remove(&scope_id);
1255            shrink_watchers_if_sparse(&mut watchers);
1256            true
1257        } else {
1258            false
1259        };
1260        drop(watchers);
1261        self.state.remove_scope_observers(usize::from(removed));
1262    }
1263
1264    fn state_id(&self) -> Option<StateId> {
1265        self.state_id.get()
1266    }
1267
1268    fn invalidate_watchers(&self) {
1269        let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1270            let mut watchers = self.watchers.borrow_mut();
1271            let before = watchers.len();
1272            let mut live = Vec::with_capacity(watchers.len());
1273            watchers.retain(|_, scope| {
1274                if let Some(inner) = scope.upgrade() {
1275                    live.push(RecomposeScope { inner });
1276                    true
1277                } else {
1278                    false
1279                }
1280            });
1281            let removed_count = before - watchers.len();
1282            shrink_watchers_if_sparse(&mut watchers);
1283            (live, removed_count)
1284        };
1285        self.state.remove_scope_observers(removed_count);
1286
1287        for watcher in watchers {
1288            debug_record_scope_invalidation::<T>(watcher.id(), self.state_id.get());
1289            if let Some(state_id) = self.state_id.get() {
1290                watcher.invalidate_from_state(state_id);
1291            } else {
1292                watcher.invalidate();
1293            }
1294        }
1295    }
1296}
1297
1298impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1299    fn drop(&mut self) {
1300        self.state
1301            .remove_scope_observers(self.watchers.get_mut().len());
1302    }
1303}
1304
1305fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1306    let Some(Some(scope)) =
1307        with_current_composer_opt(|composer| composer.current_state_invalidation_scope())
1308    else {
1309        return;
1310    };
1311    let (registered, became_subscribed) = inner.register_scope(&scope);
1312    if registered {
1313        if let Some(state_id) = inner.state_id() {
1314            scope.record_state_subscription(state_id);
1315        }
1316        if became_subscribed {
1317            inner.state.notify_subscribers();
1318        }
1319    }
1320}
1321
1322trait StateArenaHandle<T: Clone + 'static> {
1323    fn state_id(&self) -> StateId;
1324    fn runtime_id(&self) -> runtime::RuntimeId;
1325
1326    fn runtime_handle(&self) -> RuntimeHandle {
1327        runtime::runtime_handle_by_id(self.runtime_id())
1328            .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1329    }
1330
1331    fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1332        runtime::runtime_handle_by_id(self.runtime_id())
1333    }
1334
1335    fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1336        self.runtime_handle()
1337            .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1338    }
1339
1340    fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1341        self.runtime_handle_opt()?
1342            .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1343    }
1344}
1345
1346/// Cheap copyable read-only view of a state cell.
1347pub struct State<T: Clone + 'static> {
1348    id: StateId,
1349    runtime_id: runtime::RuntimeId,
1350    _marker: PhantomData<fn() -> T>,
1351}
1352
1353/// Cheap copyable mutable view of a state cell.
1354///
1355/// Ownership lives elsewhere: a composition slot, an [`OwnedMutableState`], or
1356/// the runtime for states created with [`crate::mutableStateOf`] /
1357/// [`MutableState::with_runtime`].
1358pub struct MutableState<T: Clone + 'static> {
1359    id: StateId,
1360    runtime_id: runtime::RuntimeId,
1361    _marker: PhantomData<fn() -> T>,
1362}
1363
1364/// Owning state handle for reclaimable state cells.
1365#[derive(Clone)]
1366pub struct OwnedMutableState<T: Clone + 'static> {
1367    state: MutableState<T>,
1368    _lease: Rc<runtime::StateHandleLease>,
1369    _marker: PhantomData<fn() -> T>,
1370}
1371
1372impl<T: Clone + 'static> PartialEq for State<T> {
1373    fn eq(&self, other: &Self) -> bool {
1374        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1375    }
1376}
1377
1378impl<T: Clone + 'static> Eq for State<T> {}
1379
1380impl<T: Clone + 'static> PartialEq for MutableState<T> {
1381    fn eq(&self, other: &Self) -> bool {
1382        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1383    }
1384}
1385
1386impl<T: Clone + 'static> Eq for MutableState<T> {}
1387
1388impl<T: Clone + 'static> Copy for State<T> {}
1389
1390impl<T: Clone + 'static> Clone for State<T> {
1391    fn clone(&self) -> Self {
1392        *self
1393    }
1394}
1395
1396impl<T: Clone + 'static> Copy for MutableState<T> {}
1397
1398impl<T: Clone + 'static> Clone for MutableState<T> {
1399    fn clone(&self) -> Self {
1400        *self
1401    }
1402}
1403
1404impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1405    fn state_id(&self) -> StateId {
1406        self.id
1407    }
1408
1409    fn runtime_id(&self) -> runtime::RuntimeId {
1410        self.runtime_id
1411    }
1412}
1413
1414impl<T: Clone + 'static> State<T> {
1415    fn subscribe_current_scope(&self) {
1416        self.with_inner(register_current_state_scope::<T>);
1417    }
1418
1419    pub fn is_alive(&self) -> bool {
1420        self.try_with_inner(|_| ()).is_some()
1421    }
1422
1423    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1424        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1425    }
1426
1427    pub fn try_value(&self) -> Option<T> {
1428        self.try_with_inner(|inner| inner.state.try_get())?
1429    }
1430
1431    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1432        let value = self.with_inner(|inner| inner.state.get());
1433        self.subscribe_current_scope();
1434        f(&value)
1435    }
1436
1437    pub fn value(&self) -> T {
1438        let value = self.with_inner(|inner| inner.state.get());
1439        self.subscribe_current_scope();
1440        value
1441    }
1442
1443    pub fn get(&self) -> T {
1444        self.value()
1445    }
1446
1447    pub fn has_subscribers(&self) -> bool {
1448        self.with_inner(MutableStateInner::has_subscribers)
1449    }
1450
1451    pub fn on_subscriber(&self, callback: Rc<dyn Fn()>) {
1452        self.with_inner(|inner| {
1453            inner
1454                .state
1455                .subscriber_callback(callback, inner.has_subscribers())
1456        });
1457    }
1458
1459    /// Counts this state as subscribed for as long as the returned guard
1460    /// lives, without observing it from any scope.
1461    ///
1462    /// Snapshot observers subscribe reads made under composition or the
1463    /// draw phase, and producers gated on [`State::has_subscribers`] — an
1464    /// infinite transition, for one — stop themselves when the last such
1465    /// subscriber leaves. A consumer that reads the value from a polling
1466    /// context (a frame effect stepping it into a derived state, a
1467    /// background task) is invisible to that accounting and would starve
1468    /// the producer it depends on. Holding this guard keeps the producer
1469    /// alive; dropping it releases the count.
1470    pub fn subscription_hold(&self) -> StateSubscriptionHold {
1471        StateSubscriptionHold {
1472            _lease: self
1473                .try_with_inner(|inner| inner.state.observation_lease())
1474                .flatten(),
1475        }
1476    }
1477}
1478
1479/// Keeps a [`State`] counted as subscribed while alive; see
1480/// [`State::subscription_hold`].
1481pub struct StateSubscriptionHold {
1482    _lease: Option<Box<dyn Any>>,
1483}
1484
1485impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1486    fn state_id(&self) -> StateId {
1487        self.id
1488    }
1489
1490    fn runtime_id(&self) -> runtime::RuntimeId {
1491        self.runtime_id
1492    }
1493}
1494
1495impl<T: Clone + 'static> MutableState<T> {
1496    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1497        runtime.alloc_persistent_state(value)
1498    }
1499
1500    fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1501        Self {
1502            id,
1503            runtime_id,
1504            _marker: PhantomData,
1505        }
1506    }
1507
1508    pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1509        Self::from_parts(lease.id(), lease.runtime().id())
1510    }
1511
1512    pub fn is_alive(&self) -> bool {
1513        self.try_with_inner(|_| ()).is_some()
1514    }
1515
1516    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1517        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1518    }
1519
1520    pub fn try_value(&self) -> Option<T> {
1521        self.try_with_inner(|inner| inner.state.try_get())?
1522    }
1523
1524    pub fn as_state(&self) -> State<T> {
1525        State {
1526            id: self.id,
1527            runtime_id: self.runtime_id,
1528            _marker: PhantomData,
1529        }
1530    }
1531
1532    pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1533        let lease = self
1534            .runtime_handle_opt()?
1535            .retain_state_lease(self.state_id())?;
1536        Some(OwnedMutableState {
1537            state: *self,
1538            _lease: lease,
1539            _marker: PhantomData,
1540        })
1541    }
1542
1543    pub fn retain(&self) -> OwnedMutableState<T> {
1544        self.try_retain()
1545            .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1546    }
1547
1548    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1549        let value = self.with_inner(|inner| inner.state.get());
1550        self.subscribe_current_scope();
1551        f(&value)
1552    }
1553
1554    pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1555        let runtime = self.runtime_handle();
1556        runtime.assert_ui_thread();
1557        runtime.with_state_arena(|arena| {
1558            arena.with_typed::<T, R>(self.state_id(), |inner| {
1559                let mut value = inner.state.get();
1560                let tracker = UpdateScope::new(inner.state.id());
1561                let result = f(&mut value);
1562                let wrote_elsewhere = tracker.finish();
1563                if !wrote_elsewhere && inner.state.set(value) {
1564                    inner.invalidate_watchers();
1565                }
1566                result
1567            })
1568        })
1569    }
1570
1571    pub fn replace(&self, value: T) {
1572        let Some(runtime) = self.runtime_handle_opt() else {
1573            log::debug!(
1574                "MutableState::replace skipped: runtime {:?} dropped",
1575                self.runtime_id()
1576            );
1577            return;
1578        };
1579        runtime.assert_ui_thread();
1580        let replaced = runtime
1581            .try_with_state_arena(|arena| {
1582                arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1583                    if inner.state.set(value) {
1584                        inner.invalidate_watchers();
1585                    }
1586                })
1587            })
1588            .flatten();
1589        if replaced.is_none() {
1590            log::debug!(
1591                "MutableState::replace skipped: state cell released (slot={}, gen={})",
1592                self.state_id().slot(),
1593                self.state_id().generation(),
1594            );
1595        }
1596    }
1597
1598    pub fn set_value(&self, value: T) {
1599        self.replace(value);
1600    }
1601
1602    pub fn set(&self, value: T) {
1603        self.replace(value);
1604    }
1605
1606    pub fn value(&self) -> T {
1607        let value = self.with_inner(|inner| inner.state.get());
1608        self.subscribe_current_scope();
1609        value
1610    }
1611
1612    pub fn get(&self) -> T {
1613        self.value()
1614    }
1615
1616    pub fn get_non_reactive(&self) -> T {
1617        self.with_inner(|inner| inner.state.get())
1618    }
1619
1620    #[doc(hidden)]
1621    pub fn runtime_state_id(&self) -> StateId {
1622        self.state_id()
1623    }
1624
1625    #[doc(hidden)]
1626    pub fn subscribe_current_scope_only(&self) {
1627        self.subscribe_current_scope();
1628    }
1629
1630    fn subscribe_current_scope(&self) {
1631        self.with_inner(register_current_state_scope::<T>);
1632    }
1633
1634    #[cfg(test)]
1635    pub(crate) fn watcher_count(&self) -> usize {
1636        self.with_inner(|inner| inner.watchers.borrow().len())
1637    }
1638
1639    #[cfg(test)]
1640    pub(crate) fn watcher_capacity(&self) -> usize {
1641        self.with_inner(|inner| inner.watchers.borrow().capacity())
1642    }
1643
1644    #[cfg(test)]
1645    pub(crate) fn subscriber_callback_count(&self) -> usize {
1646        self.with_inner(|inner| inner.state.subscriber_callback_count())
1647    }
1648
1649    #[cfg(test)]
1650    pub(crate) fn state_id_for_test(&self) -> StateId {
1651        self.state_id()
1652    }
1653
1654    #[cfg(test)]
1655    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1656        self.as_state().subscribe_scope_for_test(scope);
1657    }
1658}
1659
1660impl<T: Clone + 'static> OwnedMutableState<T> {
1661    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1662        let lease = runtime.alloc_state(value);
1663        Self {
1664            state: MutableState::from_lease(&lease),
1665            _lease: lease,
1666            _marker: PhantomData,
1667        }
1668    }
1669
1670    pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1671    where
1672        T: PartialEq,
1673    {
1674        Self::with_runtime_and_policy(value, runtime, Arc::new(StructuralEqual))
1675    }
1676
1677    pub(crate) fn with_runtime_and_policy(
1678        value: T,
1679        runtime: RuntimeHandle,
1680        policy: Arc<dyn MutationPolicy<T>>,
1681    ) -> Self {
1682        let lease = runtime.alloc_state_with_policy(value, policy);
1683        Self {
1684            state: MutableState::from_lease(&lease),
1685            _lease: lease,
1686            _marker: PhantomData,
1687        }
1688    }
1689
1690    pub fn handle(&self) -> MutableState<T> {
1691        self.state
1692    }
1693
1694    pub fn as_state(&self) -> State<T> {
1695        self.state.as_state()
1696    }
1697}
1698
1699impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1700    type Target = MutableState<T>;
1701
1702    fn deref(&self) -> &Self::Target {
1703        &self.state
1704    }
1705}
1706
1707#[cfg(test)]
1708impl<T: Clone + 'static> State<T> {
1709    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1710        self.with_inner(|inner| {
1711            let (registered, became_subscribed) = inner.register_scope(scope);
1712            if registered {
1713                if let Some(state_id) = inner.state_id() {
1714                    scope.record_state_subscription(state_id);
1715                }
1716                if became_subscribed {
1717                    inner.state.notify_subscribers();
1718                }
1719            }
1720        });
1721    }
1722}
1723
1724impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1725    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1726        if let Some(value) = self.try_value() {
1727            f.debug_struct("MutableState")
1728                .field("value", &value)
1729                .finish()
1730        } else {
1731            f.write_str("MutableState { value: <unavailable> }")
1732        }
1733    }
1734}
1735
1736#[derive(Clone)]
1737pub struct SnapshotStateList<T: Clone + 'static> {
1738    state: OwnedMutableState<Vec<T>>,
1739}
1740
1741impl<T: Clone + 'static> SnapshotStateList<T> {
1742    pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1743    where
1744        I: IntoIterator<Item = T>,
1745    {
1746        let initial: Vec<T> = values.into_iter().collect();
1747        Self {
1748            state: OwnedMutableState::with_runtime(initial, runtime),
1749        }
1750    }
1751
1752    pub fn as_state(&self) -> State<Vec<T>> {
1753        self.state.as_state()
1754    }
1755
1756    pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1757        self.state.handle()
1758    }
1759
1760    pub fn len(&self) -> usize {
1761        self.state.with(|values| values.len())
1762    }
1763
1764    pub fn is_empty(&self) -> bool {
1765        self.len() == 0
1766    }
1767
1768    pub fn to_vec(&self) -> Vec<T> {
1769        self.state.with(|values| values.clone())
1770    }
1771
1772    pub fn iter(&self) -> Vec<T> {
1773        self.to_vec()
1774    }
1775
1776    pub fn get(&self, index: usize) -> T {
1777        self.state.with(|values| values[index].clone())
1778    }
1779
1780    pub fn get_opt(&self, index: usize) -> Option<T> {
1781        self.state.with(|values| values.get(index).cloned())
1782    }
1783
1784    pub fn first(&self) -> Option<T> {
1785        self.get_opt(0)
1786    }
1787
1788    pub fn last(&self) -> Option<T> {
1789        self.state.with(|values| values.last().cloned())
1790    }
1791
1792    pub fn push(&self, value: T) {
1793        self.state.update(|values| values.push(value));
1794    }
1795
1796    pub fn extend<I>(&self, iter: I)
1797    where
1798        I: IntoIterator<Item = T>,
1799    {
1800        self.state.update(|values| values.extend(iter));
1801    }
1802
1803    pub fn insert(&self, index: usize, value: T) {
1804        self.state.update(|values| values.insert(index, value));
1805    }
1806
1807    pub fn set(&self, index: usize, value: T) -> T {
1808        self.state
1809            .update(|values| std::mem::replace(&mut values[index], value))
1810    }
1811
1812    pub fn remove(&self, index: usize) -> T {
1813        self.state.update(|values| values.remove(index))
1814    }
1815
1816    pub fn pop(&self) -> Option<T> {
1817        self.state.update(|values| values.pop())
1818    }
1819
1820    pub fn clear(&self) {
1821        self.state.replace(Vec::new());
1822    }
1823
1824    pub fn retain<F>(&self, mut predicate: F)
1825    where
1826        F: FnMut(&T) -> bool,
1827    {
1828        self.state
1829            .update(|values| values.retain(|value| predicate(value)));
1830    }
1831
1832    pub fn replace_with<I>(&self, iter: I)
1833    where
1834        I: IntoIterator<Item = T>,
1835    {
1836        self.state.replace(iter.into_iter().collect());
1837    }
1838}
1839
1840impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
1841    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1842        let contents = self.to_vec();
1843        f.debug_struct("SnapshotStateList")
1844            .field("values", &contents)
1845            .finish()
1846    }
1847}
1848
1849#[derive(Clone)]
1850pub struct SnapshotStateMap<K, V>
1851where
1852    K: Clone + Eq + Hash + 'static,
1853    V: Clone + 'static,
1854{
1855    state: OwnedMutableState<HashMap<K, V>>,
1856}
1857
1858impl<K, V> SnapshotStateMap<K, V>
1859where
1860    K: Clone + Eq + Hash + 'static,
1861    V: Clone + 'static,
1862{
1863    pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
1864    where
1865        I: IntoIterator<Item = (K, V)>,
1866    {
1867        let map: HashMap<K, V> = pairs.into_iter().collect();
1868        Self {
1869            state: OwnedMutableState::with_runtime(map, runtime),
1870        }
1871    }
1872
1873    pub fn as_state(&self) -> State<HashMap<K, V>> {
1874        self.state.as_state()
1875    }
1876
1877    pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
1878        self.state.handle()
1879    }
1880
1881    pub fn len(&self) -> usize {
1882        self.state.with(|map| map.len())
1883    }
1884
1885    pub fn is_empty(&self) -> bool {
1886        self.state.with(|map| map.is_empty())
1887    }
1888
1889    pub fn contains_key(&self, key: &K) -> bool {
1890        self.state.with(|map| map.contains_key(key))
1891    }
1892
1893    pub fn get(&self, key: &K) -> Option<V> {
1894        self.state.with(|map| map.get(key).cloned())
1895    }
1896
1897    pub fn to_hash_map(&self) -> HashMap<K, V> {
1898        self.state.with(|map| map.clone())
1899    }
1900
1901    pub fn insert(&self, key: K, value: V) -> Option<V> {
1902        self.state.update(|map| map.insert(key, value))
1903    }
1904
1905    pub fn extend<I>(&self, iter: I)
1906    where
1907        I: IntoIterator<Item = (K, V)>,
1908    {
1909        self.state.update(|map| map.extend(iter));
1910    }
1911
1912    pub fn remove(&self, key: &K) -> Option<V> {
1913        self.state.update(|map| map.remove(key))
1914    }
1915
1916    pub fn clear(&self) {
1917        self.state.replace(HashMap::default());
1918    }
1919
1920    pub fn retain<F>(&self, mut predicate: F)
1921    where
1922        F: FnMut(&K, &mut V) -> bool,
1923    {
1924        self.state.update(|map| map.retain(|k, v| predicate(k, v)));
1925    }
1926}
1927
1928impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
1929where
1930    K: Clone + Eq + Hash + fmt::Debug + 'static,
1931    V: Clone + fmt::Debug + 'static,
1932{
1933    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1934        let contents = self.to_hash_map();
1935        f.debug_struct("SnapshotStateMap")
1936            .field("entries", &contents)
1937            .finish()
1938    }
1939}
1940
1941pub(crate) struct DerivedState<T: Clone + 'static> {
1942    compute: Rc<dyn Fn() -> T>,
1943    pub(crate) state: OwnedMutableState<T>,
1944}
1945
1946impl<T: Clone + 'static> DerivedState<T> {
1947    pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
1948        let initial = compute();
1949        Self {
1950            compute,
1951            state: OwnedMutableState::with_runtime(initial, runtime),
1952        }
1953    }
1954
1955    pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
1956        self.compute = compute;
1957    }
1958
1959    pub(crate) fn recompute(&self) {
1960        let value = (self.compute)();
1961        self.state.set_value(value);
1962    }
1963}
1964
1965impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
1966    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1967        if let Some(value) = self.try_value() {
1968            f.debug_struct("State").field("value", &value).finish()
1969        } else {
1970            f.write_str("State { value: <unavailable> }")
1971        }
1972    }
1973}
1974
1975#[cfg(test)]
1976mod tests {
1977    use super::*;
1978
1979    fn create_record_chain(ids: &[SnapshotId]) -> Rc<StateRecord> {
1980        let mut head: Option<Rc<StateRecord>> = None;
1981
1982        for &id in ids.iter().rev() {
1983            head = Some(StateRecord::new(id, 0i32, head));
1984        }
1985
1986        head.expect("create_record_chain called with empty ids")
1987    }
1988
1989    struct ManualState {
1990        head: Rc<StateRecord>,
1991    }
1992
1993    impl ManualState {
1994        fn new(head: Rc<StateRecord>) -> Self {
1995            Self { head }
1996        }
1997    }
1998
1999    impl StateObject for ManualState {
2000        fn object_id(&self) -> ObjectId {
2001            ObjectId(999)
2002        }
2003
2004        fn first_record(&self) -> Rc<StateRecord> {
2005            Rc::clone(&self.head)
2006        }
2007
2008        fn try_readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Option<Rc<StateRecord>> {
2009            Some(Rc::clone(&self.head))
2010        }
2011
2012        fn readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Rc<StateRecord> {
2013            Rc::clone(&self.head)
2014        }
2015
2016        fn prepend_state_record(&self, _: Rc<StateRecord>) {}
2017
2018        fn promote_record(&self, _: SnapshotId) -> Result<(), &'static str> {
2019            Ok(())
2020        }
2021
2022        fn as_any(&self) -> &dyn Any {
2023            self
2024        }
2025    }
2026
2027    fn poison_mutex<T>(mutex: &Mutex<T>) {
2028        let poison_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2029            let _guard = mutex
2030                .lock()
2031                .unwrap_or_else(|poisoned| poisoned.into_inner());
2032            panic!("poison snapshot state mutex for recovery test");
2033        }));
2034
2035        assert!(poison_result.is_err());
2036    }
2037
2038    #[test]
2039    fn observation_leases_drive_subscriber_liveness() {
2040        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2041        let notifications = Rc::new(Cell::new(0));
2042        let notifications_for_callback = Rc::clone(&notifications);
2043        let callback: Rc<dyn Fn()> = Rc::new(move || {
2044            notifications_for_callback.set(notifications_for_callback.get() + 1);
2045        });
2046        state.subscriber_callback(callback.clone(), false);
2047
2048        let first = StateObject::observation_lease(&*state).expect("first observation lease");
2049        let second = StateObject::observation_lease(&*state).expect("second observation lease");
2050        assert!(state.has_subscribers());
2051        assert_eq!(notifications.get(), 1);
2052
2053        drop(first);
2054        assert!(state.has_subscribers());
2055        drop(second);
2056        assert!(!state.has_subscribers());
2057
2058        let third = StateObject::observation_lease(&*state).expect("third observation lease");
2059        assert_eq!(notifications.get(), 2);
2060        drop(third);
2061    }
2062
2063    #[test]
2064    fn snapshot_mutable_state_recovers_poisoned_weak_self_lock() {
2065        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2066
2067        poison_mutex(&state.weak_self);
2068
2069        assert_eq!(state.get(), 100);
2070        assert!(state.set(101));
2071        assert_eq!(state.get(), 101);
2072    }
2073
2074    #[test]
2075    fn snapshot_mutable_state_recovers_poisoned_apply_observer_lock() {
2076        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2077        let calls = Rc::new(Cell::new(0usize));
2078        let observed_calls = Rc::clone(&calls);
2079
2080        poison_mutex(&state.apply_observers);
2081
2082        state.add_apply_observer(Box::new(move || {
2083            observed_calls.set(observed_calls.get() + 1);
2084        }));
2085        state.notify_applied();
2086
2087        assert_eq!(calls.get(), 1);
2088    }
2089
2090    #[test]
2091    fn snapshot_mutable_state_promote_missing_record_returns_error() {
2092        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2093        let missing_snapshot = usize::MAX - 17;
2094
2095        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2096            StateObject::promote_record(&*state, missing_snapshot)
2097        }));
2098
2099        assert!(
2100            matches!(result, Ok(Err("missing child record"))),
2101            "missing child record should be reported through Result, got {result:?}"
2102        );
2103    }
2104
2105    #[test]
2106    fn snapshot_mutable_state_promote_wrong_record_type_returns_error() {
2107        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2108        let child_snapshot = usize::MAX - 31;
2109        let wrong_record = StateRecord::new(child_snapshot, "wrong type", None);
2110        StateObject::prepend_state_record(&*state, wrong_record);
2111
2112        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2113            StateObject::promote_record(&*state, child_snapshot)
2114        }));
2115
2116        assert!(
2117            matches!(result, Ok(Err("child record value missing or wrong type"))),
2118            "wrong child record type should be reported through Result, got {result:?}"
2119        );
2120    }
2121
2122    #[test]
2123    fn snapshot_mutable_state_commit_wrong_record_type_returns_error() {
2124        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2125        let merged = StateRecord::new(usize::MAX - 43, "wrong type", None);
2126
2127        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2128            StateObject::commit_merged_record(&*state, merged)
2129        }));
2130
2131        assert!(
2132            matches!(result, Ok(Err("merged record value missing or wrong type"))),
2133            "wrong merged record type should be reported through Result, got {result:?}"
2134        );
2135    }
2136
2137    #[test]
2138    fn snapshot_mutable_state_merge_wrong_record_type_returns_none() {
2139        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2140        let previous = StateRecord::new(usize::MAX - 51, 1i32, None);
2141        let current = StateRecord::new(usize::MAX - 52, "wrong type", None);
2142        let applied = StateRecord::new(usize::MAX - 53, 2i32, None);
2143
2144        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2145            StateObject::merge_records(&*state, previous, current, applied)
2146        }));
2147
2148        match result {
2149            Ok(None) => {}
2150            Ok(Some(_)) => panic!("wrong merge record type unexpectedly produced a merged record"),
2151            Err(_) => panic!("wrong merge record type should not panic"),
2152        }
2153    }
2154
2155    #[test]
2156    fn test_used_locked_finds_invalid_snapshot() {
2157        let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2158        let invalid_rec = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, Some(tail));
2159        let head = StateRecord::new(10, 0i32, Some(invalid_rec.clone()));
2160
2161        let result = used_locked(&head);
2162        assert!(result.is_some());
2163        assert_eq!(result.unwrap().snapshot_id(), INVALID_SNAPSHOT_ID);
2164    }
2165
2166    #[test]
2167    fn test_used_locked_finds_obscured_record() {
2168        crate::snapshot_pinning::reset_pinning_table();
2169
2170        let pin_handle = crate::snapshot_pinning::track_pinning(10, &SnapshotIdSet::EMPTY);
2171
2172        let oldest = StateRecord::new(2, 0i32, None);
2173        let newer = StateRecord::new(5, 0i32, Some(oldest.clone()));
2174        let head = StateRecord::new(100, 0i32, Some(newer));
2175
2176        let result = used_locked(&head);
2177
2178        assert!(result.is_some());
2179        let reused = result.unwrap();
2180        assert_eq!(
2181            reused.snapshot_id(),
2182            2,
2183            "Should return the oldest obscured record"
2184        );
2185
2186        crate::snapshot_pinning::release_pinning(pin_handle);
2187    }
2188
2189    #[test]
2190    fn test_used_locked_no_reusable_record() {
2191        crate::snapshot_pinning::reset_pinning_table();
2192
2193        let high_id = allocate_record_id() + 1000;
2194        let head = create_record_chain(&[high_id, high_id + 1, high_id + 2]);
2195
2196        let result = used_locked(&head);
2197        assert!(
2198            result.is_none(),
2199            "Should find no reusable records when all are recent"
2200        );
2201    }
2202
2203    #[test]
2204    fn test_used_locked_single_old_record() {
2205        crate::snapshot_pinning::reset_pinning_table();
2206
2207        let old = StateRecord::new(2, 0i32, None);
2208        let head = StateRecord::new(100, 0i32, Some(old));
2209
2210        let result = used_locked(&head);
2211        assert!(result.is_none(), "Single old record should not be reused");
2212    }
2213
2214    #[test]
2215    fn test_readable_record_for_preexisting() {
2216        let head = create_record_chain(&[PREEXISTING_SNAPSHOT_ID]);
2217        let invalid = SnapshotIdSet::EMPTY;
2218
2219        let result = readable_record_for(&head, 10, &invalid);
2220        assert!(result.is_some());
2221        assert_eq!(result.unwrap().snapshot_id(), PREEXISTING_SNAPSHOT_ID);
2222    }
2223
2224    #[test]
2225    fn test_readable_record_for_picks_highest_valid() {
2226        let head = create_record_chain(&[10, 5, PREEXISTING_SNAPSHOT_ID]);
2227        let invalid = SnapshotIdSet::EMPTY;
2228
2229        let result = readable_record_for(&head, 10, &invalid);
2230        assert!(result.is_some());
2231        assert_eq!(result.unwrap().snapshot_id(), 10);
2232
2233        let result = readable_record_for(&head, 7, &invalid);
2234        assert!(result.is_some());
2235        assert_eq!(result.unwrap().snapshot_id(), 5);
2236    }
2237
2238    #[test]
2239    fn test_new_overwritable_record_locked_reuses_invalid() {
2240        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2241
2242        let current_head = state.first_record();
2243        let invalid_rec = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, current_head.next());
2244        current_head.set_next(Some(invalid_rec.clone()));
2245
2246        let result = new_overwritable_record_locked(&*state);
2247
2248        assert!(Rc::ptr_eq(&result, &invalid_rec));
2249        assert_eq!(result.snapshot_id(), SNAPSHOT_ID_MAX);
2250    }
2251
2252    #[test]
2253    fn test_new_overwritable_record_locked_creates_new() {
2254        crate::snapshot_pinning::reset_pinning_table();
2255
2256        let _pin_handle = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2257
2258        let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2259        let old_head = state.first_record();
2260
2261        let result = new_overwritable_record_locked(&*state);
2262
2263        assert_eq!(result.snapshot_id(), SNAPSHOT_ID_MAX);
2264
2265        let new_head = state.first_record();
2266        assert!(
2267            Rc::ptr_eq(&new_head, &result),
2268            "new_head ({:p}) should equal result ({:p})",
2269            Rc::as_ptr(&new_head),
2270            Rc::as_ptr(&result)
2271        );
2272
2273        assert!(result.next().is_some());
2274        assert!(Rc::ptr_eq(&result.next().unwrap(), &old_head));
2275    }
2276
2277    #[test]
2278    fn test_writable_record_reuses_invalid_record() {
2279        crate::snapshot_pinning::reset_pinning_table();
2280
2281        let state = SnapshotMutableState::new_in_arc(7i32, Arc::new(NeverEqual));
2282
2283        let head = state.first_record();
2284        let invalid = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, head.next());
2285        head.set_next(Some(invalid.clone()));
2286
2287        let snapshot_id = allocate_record_id();
2288        let result = state.writable_record(snapshot_id, &SnapshotIdSet::EMPTY);
2289
2290        assert!(
2291            Rc::ptr_eq(&result, &invalid),
2292            "Expected writable_record to reuse the INVALID record"
2293        );
2294        assert_eq!(result.snapshot_id(), snapshot_id);
2295        result.with_value(|value: &i32| {
2296            assert_eq!(*value, 7, "Reused record should copy the readable value");
2297        });
2298        assert!(!result.is_tombstone());
2299    }
2300
2301    #[test]
2302    fn test_writable_record_creates_new_when_reuse_disallowed() {
2303        crate::snapshot_pinning::reset_pinning_table();
2304        let pin = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2305
2306        let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2307        let original_head = state.first_record();
2308        let preexisting = original_head
2309            .next()
2310            .expect("preexisting record should exist for newly created state");
2311
2312        let snapshot_id = allocate_record_id();
2313        let result = state.writable_record(snapshot_id, &SnapshotIdSet::EMPTY);
2314
2315        assert!(
2316            !Rc::ptr_eq(&result, &original_head),
2317            "Should not reuse the current head when reuse is disallowed"
2318        );
2319        assert!(
2320            !Rc::ptr_eq(&result, &preexisting),
2321            "Should not reuse the PREEXISTING record"
2322        );
2323        assert_eq!(result.snapshot_id(), snapshot_id);
2324        result.with_value(|value: &i32| assert_eq!(*value, 42));
2325
2326        let new_head = state.first_record();
2327        assert!(
2328            Rc::ptr_eq(&new_head, &result),
2329            "Newly created record should become the head of the chain"
2330        );
2331
2332        crate::snapshot_pinning::release_pinning(pin);
2333    }
2334
2335    #[test]
2336    fn test_state_record_clear_for_reuse() {
2337        let record = StateRecord::new(10, 42i32, None);
2338
2339        record.with_value(|val: &i32| {
2340            assert_eq!(*val, 42);
2341        });
2342
2343        record.clear_for_reuse();
2344
2345        assert_eq!(record.snapshot_id(), 10);
2346    }
2347
2348    #[test]
2349    fn test_overwrite_unused_records_no_old_records() {
2350        crate::snapshot_pinning::reset_pinning_table();
2351
2352        let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2353
2354        let _pin = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2355
2356        let should_retain = state.overwrite_unused_records();
2357
2358        assert!(
2359            should_retain,
2360            "Should retain multiple records when none are old enough"
2361        );
2362
2363        let mut cursor = Some(state.first_record());
2364        while let Some(record) = cursor {
2365            assert_ne!(record.snapshot_id(), INVALID_SNAPSHOT_ID);
2366            cursor = record.next();
2367        }
2368    }
2369
2370    #[test]
2371    fn test_overwrite_unused_records_basic_cleanup() {
2372        crate::snapshot_pinning::reset_pinning_table();
2373
2374        let rec1 = StateRecord::new(100, 1i32, None);
2375        let rec2 = StateRecord::new(200, 2i32, Some(rec1.clone()));
2376        let rec3 = StateRecord::new(300, 3i32, Some(rec2.clone()));
2377
2378        let test_state = ManualState::new(rec3.clone());
2379
2380        let _pin = crate::snapshot_pinning::track_pinning(1000, &SnapshotIdSet::EMPTY);
2381
2382        let result = overwrite_unused_records_locked::<i32>(&test_state);
2383
2384        assert_eq!(rec3.snapshot_id(), 300);
2385        assert_eq!(rec2.snapshot_id(), INVALID_SNAPSHOT_ID);
2386        assert_eq!(rec1.snapshot_id(), INVALID_SNAPSHOT_ID);
2387
2388        assert!(!result);
2389    }
2390
2391    #[test]
2392    fn test_overwrite_unused_records_single_record_only() {
2393        crate::snapshot_pinning::reset_pinning_table();
2394
2395        let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2396
2397        let head = state.first_record();
2398        head.set_next(None);
2399
2400        let should_retain = state.overwrite_unused_records();
2401
2402        assert!(!should_retain, "Single record should return false");
2403    }
2404
2405    #[test]
2406    fn snapshot_state_try_get_reports_missing_visible_record_without_panicking() {
2407        crate::snapshot_pinning::reset_pinning_table();
2408
2409        let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2410        let head = state.first_record();
2411        head.set_snapshot_id(SNAPSHOT_ID_MAX);
2412        head.set_next(None);
2413
2414        assert_eq!(state.try_get(), None);
2415    }
2416
2417    #[test]
2418    fn test_overwrite_unused_records_clears_values() {
2419        crate::snapshot_pinning::reset_pinning_table();
2420
2421        let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2422        let old_rec1 = StateRecord::new(2, 999i32, Some(tail.clone()));
2423        let old_rec2 = StateRecord::new(3, 888i32, Some(old_rec1.clone()));
2424        let head = StateRecord::new(150, 42i32, Some(old_rec2.clone()));
2425        let state = ManualState::new(head.clone());
2426
2427        old_rec1.with_value(|val: &i32| {
2428            assert_eq!(*val, 999);
2429        });
2430
2431        let _pin = crate::snapshot_pinning::track_pinning(100, &SnapshotIdSet::EMPTY);
2432        overwrite_unused_records_locked::<i32>(&state);
2433
2434        assert_eq!(old_rec1.snapshot_id(), INVALID_SNAPSHOT_ID);
2435    }
2436
2437    #[test]
2438    fn test_overwrite_unused_records_mixed_old_and_new() {
2439        crate::snapshot_pinning::reset_pinning_table();
2440
2441        let preexisting = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2442        let rec2 = StateRecord::new(2, 100i32, Some(preexisting.clone()));
2443        let rec5 = StateRecord::new(5, 100i32, Some(rec2.clone()));
2444        let rec50 = StateRecord::new(50, 100i32, Some(rec5.clone()));
2445        let head = StateRecord::new(120, 100i32, Some(rec50.clone()));
2446        let state = ManualState::new(head.clone());
2447
2448        let _pin = crate::snapshot_pinning::track_pinning(40, &SnapshotIdSet::EMPTY);
2449
2450        let should_retain = overwrite_unused_records_locked::<i32>(&state);
2451        assert!(should_retain);
2452
2453        assert_eq!(rec50.snapshot_id(), 50);
2454        assert_eq!(rec5.snapshot_id(), 5);
2455        assert_eq!(rec2.snapshot_id(), INVALID_SNAPSHOT_ID);
2456    }
2457
2458    #[test]
2459    fn test_readable_record_for_skips_invalid_set() {
2460        let head = create_record_chain(&[10, 5, PREEXISTING_SNAPSHOT_ID]);
2461        let invalid = SnapshotIdSet::new().set(5);
2462
2463        let result = readable_record_for(&head, 10, &invalid);
2464        assert!(result.is_some());
2465        assert_eq!(result.unwrap().snapshot_id(), 10);
2466
2467        let result = readable_record_for(&head, 7, &invalid);
2468        assert!(result.is_some());
2469        assert_eq!(result.unwrap().snapshot_id(), PREEXISTING_SNAPSHOT_ID);
2470    }
2471
2472    #[test]
2473    fn test_assign_value_copies_int() {
2474        let source = StateRecord::new(10, 42i32, None);
2475        let target = StateRecord::new(20, 0i32, None);
2476
2477        target.assign_value::<i32>(&source).expect("copy int value");
2478
2479        target.with_value(|val: &i32| {
2480            assert_eq!(*val, 42);
2481        });
2482
2483        source.with_value(|val: &i32| {
2484            assert_eq!(*val, 42);
2485        });
2486
2487        assert_eq!(source.snapshot_id(), 10);
2488        assert_eq!(target.snapshot_id(), 20);
2489    }
2490
2491    #[test]
2492    fn test_assign_value_copies_string() {
2493        let source = StateRecord::new(10, "hello".to_string(), None);
2494        let target = StateRecord::new(20, "world".to_string(), None);
2495
2496        target
2497            .assign_value::<String>(&source)
2498            .expect("copy string value");
2499
2500        target.with_value(|val: &String| {
2501            assert_eq!(val, "hello");
2502        });
2503
2504        source.with_value(|val: &String| {
2505            assert_eq!(val, "hello");
2506        });
2507    }
2508
2509    #[test]
2510    fn test_assign_value_reports_cleared_source() {
2511        let source = StateRecord::new(10, 42i32, None);
2512        let target = StateRecord::new(20, 0i32, None);
2513
2514        source.clear_value();
2515
2516        assert_eq!(
2517            target.assign_value::<i32>(&source),
2518            Err(StateRecordValueError::MissingOrWrongType {
2519                expected: std::any::type_name::<i32>(),
2520            })
2521        );
2522        assert_eq!(target.with_value(|val: &i32| *val), 0);
2523    }
2524
2525    #[test]
2526    fn test_assign_value_overwrites_existing_value() {
2527        let source = StateRecord::new(10, 100i32, None);
2528        let target = StateRecord::new(20, 999i32, None);
2529
2530        target.with_value(|val: &i32| {
2531            assert_eq!(*val, 999);
2532        });
2533
2534        target
2535            .assign_value::<i32>(&source)
2536            .expect("overwrite int value");
2537
2538        target.with_value(|val: &i32| {
2539            assert_eq!(*val, 100);
2540        });
2541    }
2542
2543    #[test]
2544    fn test_assign_value_with_custom_type() {
2545        #[derive(Clone, PartialEq, Debug)]
2546        struct Point {
2547            x: f64,
2548            y: f64,
2549        }
2550
2551        let source = StateRecord::new(10, Point { x: 1.5, y: 2.5 }, None);
2552        let target = StateRecord::new(20, Point { x: 0.0, y: 0.0 }, None);
2553
2554        target
2555            .assign_value::<Point>(&source)
2556            .expect("copy point value");
2557
2558        target.with_value(|val: &Point| {
2559            assert_eq!(val, &Point { x: 1.5, y: 2.5 });
2560        });
2561    }
2562
2563    #[test]
2564    fn test_assign_value_self_assignment() {
2565        let record = StateRecord::new(10, 42i32, None);
2566
2567        record
2568            .assign_value::<i32>(&record)
2569            .expect("self-assign int value");
2570
2571        record.with_value(|val: &i32| {
2572            assert_eq!(*val, 42);
2573        });
2574    }
2575
2576    #[test]
2577    fn event_loop_writes_keep_the_record_chain_bounded() {
2578        crate::snapshot_pinning::reset_pinning_table();
2579        let state = SnapshotMutableState::new_in_arc(0.0f32, Arc::new(NeverEqual));
2580
2581        let mut lens = Vec::new();
2582        for event in 0..3000usize {
2583            crate::run_in_mutable_snapshot(|| {
2584                state.set(event as f32);
2585            })
2586            .expect("event snapshot applies");
2587            let _ = state.get();
2588            if event % 500 == 499 {
2589                lens.push(state.record_chain_debug().len());
2590            }
2591        }
2592
2593        let final_len = *lens.last().expect("sampled chain lengths");
2594        assert!(
2595            final_len <= 16,
2596            "record chain grew without bound across event-loop writes: {lens:?}"
2597        );
2598    }
2599
2600    #[test]
2601    fn test_assign_value_with_vec() {
2602        let source = StateRecord::new(10, vec![1, 2, 3, 4, 5], None);
2603        let target = StateRecord::new(20, Vec::<i32>::new(), None);
2604
2605        target
2606            .assign_value::<Vec<i32>>(&source)
2607            .expect("copy vec value");
2608
2609        target.with_value(|val: &Vec<i32>| {
2610            assert_eq!(val, &vec![1, 2, 3, 4, 5]);
2611        });
2612
2613        source.replace_value(vec![10, 20]);
2614        target.with_value(|val: &Vec<i32>| {
2615            assert_eq!(val, &vec![1, 2, 3, 4, 5]);
2616        });
2617    }
2618}