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