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