Skip to main content

cranpose_core/
state.rs

1use std::{
2    any::Any,
3    cell::{Cell, RefCell},
4    fmt,
5    hash::Hash,
6    marker::PhantomData,
7    ops::Deref,
8    rc::{Rc, Weak as RcWeak},
9    sync::{Arc, Mutex, MutexGuard, PoisonError, Weak},
10};
11
12use crate::{
13    RecomposeScope, RecomposeScopeInner, RuntimeHandle, ScopeId, StateId,
14    collections::map::{HashMap, HashSet},
15    debug_trace::debug_record_scope_invalidation,
16    runtime,
17    snapshot_id_set::{SnapshotId, SnapshotIdSet},
18    snapshot_pinning::lowest_pinned_snapshot,
19    snapshot_v2::{
20        AnySnapshot, GlobalSnapshot, advance_global_snapshot, allocate_record_id, current_snapshot,
21    },
22    with_current_composer_opt,
23};
24
25pub(crate) const PREEXISTING_SNAPSHOT_ID: SnapshotId = 1;
26
27const INVALID_SNAPSHOT_ID: SnapshotId = 0;
28
29const SNAPSHOT_ID_MAX: SnapshotId = usize::MAX;
30
31#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug, Default)]
32pub struct ObjectId(pub(crate) usize);
33
34impl ObjectId {
35    pub(crate) fn new<T: ?Sized + 'static>(object: &Arc<T>) -> Self {
36        Self(Arc::as_ptr(object) as *const () as usize)
37    }
38
39    #[inline]
40    pub(crate) fn as_usize(self) -> usize {
41        self.0
42    }
43}
44
45pub struct StateRecord {
46    snapshot_id: Cell<SnapshotId>,
47    tombstone: Cell<bool>,
48    next: Cell<Option<Rc<StateRecord>>>,
49    value: RefCell<Option<Box<dyn Any>>>,
50}
51
52#[derive(Debug)]
53struct StateReadFailure {
54    state_id: ObjectId,
55    snapshot_id: SnapshotId,
56    fresh_snapshot_id: SnapshotId,
57    fresh_invalid: SnapshotIdSet,
58    record_chain: Vec<(SnapshotId, bool)>,
59}
60
61impl std::fmt::Display for StateReadFailure {
62    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        write!(
64            f,
65            "Reading a state that was created after the snapshot was taken or in a snapshot that has not yet been applied\n\
66             state={:?}, snapshot_id={}, fresh_snapshot_id={}, fresh_invalid={:?}\n\
67             record_chain={:?}",
68            self.state_id,
69            self.snapshot_id,
70            self.fresh_snapshot_id,
71            self.fresh_invalid,
72            self.record_chain
73        )
74    }
75}
76
77#[derive(Debug, Clone, Copy, Eq, PartialEq)]
78pub(crate) enum StateRecordValueError {
79    MissingOrWrongType { expected: &'static str },
80}
81
82impl StateRecord {
83    pub(crate) fn new<T: Any>(
84        snapshot_id: SnapshotId,
85        value: T,
86        next: Option<Rc<StateRecord>>,
87    ) -> Rc<Self> {
88        Rc::new(Self {
89            snapshot_id: Cell::new(snapshot_id),
90            tombstone: Cell::new(false),
91            next: Cell::new(next),
92            value: RefCell::new(Some(Box::new(value))),
93        })
94    }
95
96    #[inline]
97    pub(crate) fn snapshot_id(&self) -> SnapshotId {
98        self.snapshot_id.get()
99    }
100
101    #[inline]
102    pub(crate) fn set_snapshot_id(&self, id: SnapshotId) {
103        self.snapshot_id.set(id);
104    }
105
106    #[inline]
107    pub(crate) fn next(&self) -> Option<Rc<StateRecord>> {
108        self.next.take().inspect(|record| {
109            self.next.set(Some(Rc::clone(record)));
110        })
111    }
112
113    #[inline]
114    pub(crate) fn set_next(&self, next: Option<Rc<StateRecord>>) {
115        self.next.set(next);
116    }
117
118    #[inline]
119    pub(crate) fn is_tombstone(&self) -> bool {
120        self.tombstone.get()
121    }
122
123    #[inline]
124    pub(crate) fn set_tombstone(&self, tombstone: bool) {
125        self.tombstone.set(tombstone);
126    }
127
128    pub(crate) fn clear_value(&self) {
129        self.value.borrow_mut().take();
130    }
131
132    pub(crate) fn replace_value<T: Any>(&self, new_value: T) {
133        *self.value.borrow_mut() = Some(Box::new(new_value));
134    }
135
136    pub(crate) fn with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> R {
137        self.try_with_value(f)
138            .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"))
139    }
140
141    pub(crate) fn try_with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
142        let guard = self.value.borrow();
143        let value = guard.as_ref().and_then(|boxed| boxed.downcast_ref::<T>())?;
144        Some(f(value))
145    }
146
147    #[cfg(test)]
148    pub(crate) fn clear_for_reuse(&self) {
149        self.clear_value();
150    }
151
152    pub(crate) fn assign_value<T: Any + Clone>(
153        &self,
154        source: &StateRecord,
155    ) -> Result<(), StateRecordValueError> {
156        let cloned_value = source
157            .try_with_value(|value: &T| value.clone())
158            .ok_or_else(|| StateRecordValueError::MissingOrWrongType {
159                expected: std::any::type_name::<T>(),
160            })?;
161        self.replace_value(cloned_value);
162        Ok(())
163    }
164}
165
166impl Drop for StateRecord {
167    fn drop(&mut self) {
168        let mut next = self.next.take();
169        while let Some(node) = next {
170            match Rc::try_unwrap(node) {
171                Ok(record) => {
172                    next = record.next.take();
173                }
174                Err(_) => {
175                    break;
176                }
177            }
178        }
179    }
180}
181
182struct CurrentRecord {
183    head: RefCell<Rc<StateRecord>>,
184}
185
186impl CurrentRecord {
187    fn new(head: Rc<StateRecord>) -> Self {
188        Self {
189            head: RefCell::new(head),
190        }
191    }
192
193    fn clone_head(&self) -> Rc<StateRecord> {
194        self.head.borrow().clone()
195    }
196
197    fn replace(&self, new_head: Rc<StateRecord>) {
198        *self.head.borrow_mut() = new_head;
199    }
200
201    fn prepend(&self, record: Rc<StateRecord>) {
202        let current_head = self.clone_head();
203        record.set_next(Some(current_head));
204        self.replace(record);
205    }
206}
207
208#[inline]
209fn record_is_valid_for(
210    record: &Rc<StateRecord>,
211    snapshot_id: SnapshotId,
212    invalid: &SnapshotIdSet,
213) -> bool {
214    if record.is_tombstone() {
215        return false;
216    }
217
218    let candidate = record.snapshot_id();
219    if candidate == INVALID_SNAPSHOT_ID || candidate > snapshot_id {
220        return false;
221    }
222
223    candidate == snapshot_id || !invalid.get(candidate)
224}
225
226pub(crate) fn readable_record_for(
227    head: &Rc<StateRecord>,
228    snapshot_id: SnapshotId,
229    invalid: &SnapshotIdSet,
230) -> Option<Rc<StateRecord>> {
231    let mut best: Option<Rc<StateRecord>> = None;
232    let mut cursor = Some(Rc::clone(head));
233
234    while let Some(record) = cursor {
235        if record_is_valid_for(&record, snapshot_id, invalid) {
236            let replace = best
237                .as_ref()
238                .is_none_or(|current| current.snapshot_id() < record.snapshot_id());
239            if replace {
240                best = Some(Rc::clone(&record));
241            }
242        }
243        cursor = record.next();
244    }
245
246    best
247}
248
249fn find_youngest_or<F>(head: &Rc<StateRecord>, predicate: F) -> Rc<StateRecord>
250where
251    F: Fn(&Rc<StateRecord>) -> bool,
252{
253    let mut current = Some(Rc::clone(head));
254    let mut youngest = Rc::clone(head);
255
256    while let Some(record) = current {
257        if predicate(&record) {
258            return record;
259        }
260        if youngest.snapshot_id() < record.snapshot_id() {
261            youngest = Rc::clone(&record);
262        }
263        current = record.next();
264    }
265
266    youngest
267}
268
269pub(crate) fn used_locked(head: &Rc<StateRecord>) -> Option<Rc<StateRecord>> {
270    let mut current = Some(Rc::clone(head));
271    let mut valid_record: Option<Rc<StateRecord>> = None;
272
273    let reuse_limit = lowest_pinned_snapshot().map_or_else(
274        || allocate_record_id().saturating_sub(1),
275        |lowest| lowest.saturating_sub(1),
276    );
277
278    let invalid = SnapshotIdSet::EMPTY;
279
280    while let Some(record) = current {
281        let current_id = record.snapshot_id();
282
283        if current_id == PREEXISTING_SNAPSHOT_ID {
284            current = record.next();
285            continue;
286        }
287
288        if current_id == INVALID_SNAPSHOT_ID {
289            return Some(record);
290        }
291
292        if record.is_tombstone() && current_id < reuse_limit {
293            return Some(record);
294        }
295
296        if record_is_valid_for(&record, reuse_limit, &invalid) {
297            if let Some(ref existing) = valid_record {
298                return Some(if current_id < existing.snapshot_id() {
299                    record
300                } else {
301                    Rc::clone(existing)
302                });
303            } else {
304                valid_record = Some(record.clone());
305            }
306        }
307
308        current = record.next();
309    }
310
311    None
312}
313
314pub(crate) fn new_overwritable_record_locked(state: &dyn StateObject) -> Rc<StateRecord> {
315    let state_head = state.first_record();
316
317    if let Some(reusable) = used_locked(&state_head) {
318        reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
319        return reusable;
320    }
321
322    let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
323
324    state.prepend_state_record(Rc::clone(&new_record));
325
326    new_record
327}
328
329pub(crate) fn new_overwritable_record_as_head_locked(state: &dyn StateObject) -> Rc<StateRecord> {
330    let head = state.first_record();
331
332    if let Some(reusable) = used_locked(&head) {
333        reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
334
335        if !Rc::ptr_eq(&head, &reusable) {
336            let mut cursor = Some(Rc::clone(&head));
337            let mut unlinked = false;
338
339            while let Some(node) = cursor {
340                let next = node.next();
341                if let Some(next_record) = next {
342                    if Rc::ptr_eq(&next_record, &reusable) {
343                        node.set_next(reusable.next());
344                        unlinked = true;
345                        break;
346                    }
347                    cursor = Some(next_record);
348                } else {
349                    break;
350                }
351            }
352
353            if !unlinked {
354                debug_assert!(
355                    false,
356                    "new_overwritable_record_as_head_locked: reusable record not found in chain"
357                );
358                let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
359                state.prepend_state_record(Rc::clone(&new_record));
360                return new_record;
361            }
362
363            state.prepend_state_record(Rc::clone(&reusable));
364        }
365
366        return reusable;
367    }
368
369    let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
370    state.prepend_state_record(Rc::clone(&new_record));
371    new_record
372}
373
374pub(crate) fn overwrite_unused_records_locked<T: Any + Clone>(state: &dyn StateObject) -> bool {
375    let head = state.first_record();
376    let mut current = Some(Rc::clone(&head));
377    let mut overwrite_record: Option<Rc<StateRecord>> = None;
378    let mut valid_record: Option<Rc<StateRecord>> = None;
379
380    let reuse_limit =
381        lowest_pinned_snapshot().unwrap_or_else(crate::snapshot_v2::peek_next_snapshot_id);
382
383    let mut retained_records = 0;
384
385    while let Some(record) = current {
386        let current_id = record.snapshot_id();
387
388        if current_id == INVALID_SNAPSHOT_ID {
389        } else if current_id < reuse_limit {
390            if valid_record.is_none() {
391                valid_record = Some(Rc::clone(&record));
392                retained_records += 1;
393            } else {
394                let Some(valid) = valid_record.as_ref() else {
395                    valid_record = Some(Rc::clone(&record));
396                    retained_records += 1;
397                    current = record.next();
398                    continue;
399                };
400                let record_to_overwrite = if current_id < valid.snapshot_id() {
401                    Rc::clone(&record)
402                } else {
403                    let to_overwrite = Rc::clone(valid);
404                    valid_record = Some(Rc::clone(&record));
405                    to_overwrite
406                };
407
408                let source_record = overwrite_record.get_or_insert_with(|| {
409                    find_youngest_or(&head, |r| r.snapshot_id() >= reuse_limit)
410                });
411
412                record_to_overwrite.set_snapshot_id(INVALID_SNAPSHOT_ID);
413                if let Err(error) = record_to_overwrite.assign_value::<T>(source_record) {
414                    log::error!(
415                        "snapshot cleanup could not copy retained state record value for state {:?}: {:?}",
416                        state.object_id(),
417                        error
418                    );
419                }
420            }
421        } else {
422            retained_records += 1;
423        }
424
425        current = record.next();
426    }
427
428    retained_records > 1
429}
430
431fn active_snapshot() -> AnySnapshot {
432    current_snapshot().unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()))
433}
434
435pub(crate) trait MutationPolicy<T>: Send + Sync {
436    fn equivalent(&self, a: &T, b: &T) -> bool;
437    fn merge(&self, _previous: &T, _current: &T, _applied: &T) -> Option<T> {
438        None
439    }
440}
441
442pub(crate) struct NeverEqual;
443
444impl<T> MutationPolicy<T> for NeverEqual {
445    fn equivalent(&self, _a: &T, _b: &T) -> bool {
446        false
447    }
448}
449
450pub(crate) struct StructuralEqual;
451
452impl<T: PartialEq> MutationPolicy<T> for StructuralEqual {
453    fn equivalent(&self, a: &T, b: &T) -> bool {
454        a == b
455    }
456}
457
458pub trait StateObject: Any {
459    fn object_id(&self) -> ObjectId;
460    fn first_record(&self) -> Rc<StateRecord>;
461    fn try_readable_record(
462        &self,
463        snapshot_id: SnapshotId,
464        invalid: &SnapshotIdSet,
465    ) -> Option<Rc<StateRecord>>;
466    fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord>;
467
468    fn prepend_state_record(&self, record: Rc<StateRecord>);
469
470    fn merge_records(
471        &self,
472        _previous: Rc<StateRecord>,
473        _current: Rc<StateRecord>,
474        _applied: Rc<StateRecord>,
475    ) -> Option<Rc<StateRecord>> {
476        None
477    }
478
479    fn commit_merged_record(&self, _merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
480        Err("StateObject does not support merged record commits")
481    }
482    fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str>;
483
484    fn overwrite_unused_records(&self) -> bool {
485        false
486    }
487
488    fn observation_lease(&self) -> Option<Rc<dyn Any>> {
489        None
490    }
491
492    fn as_any(&self) -> &dyn Any;
493}
494
495pub(crate) struct SnapshotMutableState<T> {
496    head: CurrentRecord,
497    policy: Arc<dyn MutationPolicy<T>>,
498    id: ObjectId,
499    weak_self: Mutex<Option<Weak<Self>>>,
500    apply_observers: Mutex<Vec<Box<dyn Fn() + 'static>>>,
501    read_observation_lease: Rc<()>,
502    scope_observation_count: Cell<usize>,
503    subscriber_callbacks: RefCell<Vec<Rc<dyn Fn()>>>,
504}
505
506impl<T> SnapshotMutableState<T> {
507    fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
508        if !should_check_chain_integrity() {
509            return;
510        }
511        let head = self.head.clone_head();
512        let mut cursor = Some(head);
513        let mut seen: HashSet<usize> = HashSet::default();
514        let mut ids = Vec::new();
515
516        while let Some(record) = cursor {
517            let addr = Rc::as_ptr(&record) as usize;
518            assert!(
519                seen.insert(addr),
520                "SnapshotMutableState::{} detected duplicate/cycle at record {:p} for state {:?} (snapshot_context={:?}, chain_ids={:?})",
521                caller,
522                Rc::as_ptr(&record),
523                self.id,
524                snapshot_context,
525                ids
526            );
527            ids.push(record.snapshot_id());
528            cursor = record.next();
529        }
530
531        assert!(
532            !ids.is_empty(),
533            "SnapshotMutableState::{} finished integrity scan with empty id list for state {:?} (snapshot_context={:?})",
534            caller,
535            self.id,
536            snapshot_context
537        );
538    }
539}
540
541fn should_check_chain_integrity() -> bool {
542    #[cfg(debug_assertions)]
543    {
544        true
545    }
546
547    #[cfg(not(debug_assertions))]
548    {
549        crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
550    }
551}
552
553impl<T: Clone + 'static> SnapshotMutableState<T> {
554    fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
555        let mut chain_ids = Vec::new();
556        let mut cursor = Some(self.first_record());
557        while let Some(record) = cursor {
558            chain_ids.push((record.snapshot_id(), record.is_tombstone()));
559            cursor = record.next();
560        }
561        chain_ids
562    }
563
564    fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
565        let snapshot = active_snapshot();
566        if let Some(state) = self.upgrade_self() {
567            snapshot.record_read(&*state);
568        }
569
570        let snapshot_id = snapshot.snapshot_id();
571        let invalid = snapshot.invalid();
572
573        if let Some(record) = self.readable_for(snapshot_id, &invalid) {
574            return Ok(record);
575        }
576
577        let fresh_snapshot = active_snapshot();
578        let fresh_id = fresh_snapshot.snapshot_id();
579        let fresh_invalid = fresh_snapshot.invalid();
580
581        if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
582            return Ok(record);
583        }
584
585        let global = GlobalSnapshot::get_or_create();
586        let global_id = global.snapshot_id();
587        let global_invalid = global.invalid();
588
589        if let Some(record) = self.readable_for(global_id, &global_invalid) {
590            return Ok(record);
591        }
592
593        Err(StateReadFailure {
594            state_id: self.id,
595            snapshot_id,
596            fresh_snapshot_id: fresh_id,
597            fresh_invalid,
598            record_chain: self.record_chain_debug(),
599        })
600    }
601
602    fn readable_for(
603        &self,
604        snapshot_id: SnapshotId,
605        invalid: &SnapshotIdSet,
606    ) -> Option<Rc<StateRecord>> {
607        let head = self.first_record();
608        readable_record_for(&head, snapshot_id, invalid)
609    }
610
611    fn is_equivalent_to_readable(
612        &self,
613        snapshot_id: SnapshotId,
614        invalid: &SnapshotIdSet,
615        new_value: &T,
616    ) -> bool {
617        self.readable_for(snapshot_id, invalid)
618            .is_some_and(|record| {
619                record.with_value(|current: &T| self.policy.equivalent(current, new_value))
620            })
621    }
622
623    fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
624        let Some(readable) = self.readable_for(snapshot_id, invalid) else {
625            let current_head = self.head.clone_head();
626            let refreshed = readable_record_for(&current_head, snapshot_id, invalid);
627            let source = refreshed.unwrap_or_else(|| current_head.clone());
628
629            let cloned_value = source.with_value(|value: &T| value.clone());
630            let new_head = StateRecord::new(snapshot_id, cloned_value, Some(current_head));
631            self.head.replace(new_head.clone());
632            self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
633            return new_head;
634        };
635
636        if readable.snapshot_id() == snapshot_id {
637            return readable;
638        }
639
640        let refreshed = {
641            let current_head = self.head.clone_head();
642            let refreshed = readable_record_for(&current_head, snapshot_id, invalid).unwrap_or_else(
643                || {
644                    panic!(
645                        "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
646                        self.id, snapshot_id, invalid
647                    )
648                },
649            );
650
651            if refreshed.snapshot_id() == snapshot_id {
652                return refreshed;
653            }
654
655            Rc::clone(&refreshed)
656        };
657
658        let overwritable = new_overwritable_record_locked(self);
659        if let Err(error) = overwritable.assign_value::<T>(&refreshed) {
660            log::error!(
661                "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
662                self.id,
663                error
664            );
665        }
666        overwritable.set_snapshot_id(snapshot_id);
667        overwritable.set_tombstone(false);
668
669        self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
670
671        overwritable
672    }
673
674    pub(crate) fn new_in_arc(initial: T, policy: Arc<dyn MutationPolicy<T>>) -> Arc<Self> {
675        let snapshot = active_snapshot();
676        let snapshot_id = snapshot.snapshot_id();
677
678        let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
679        let head = StateRecord::new(snapshot_id, initial, Some(tail));
680
681        let mut state = Arc::new(Self {
682            head: CurrentRecord::new(head),
683            policy,
684            id: ObjectId::default(),
685            weak_self: Mutex::new(None),
686            apply_observers: Mutex::new(Vec::new()),
687            read_observation_lease: Rc::new(()),
688            scope_observation_count: Cell::new(0),
689            subscriber_callbacks: RefCell::new(Vec::new()),
690        });
691
692        let id = ObjectId::new(&state);
693        if let Some(state_inner) = Arc::get_mut(&mut state) {
694            state_inner.id = id;
695        }
696
697        *state.lock_weak_self() = Some(Arc::downgrade(&state));
698
699        state
700    }
701
702    pub(crate) fn add_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
703        self.lock_apply_observers().push(observer);
704    }
705
706    fn acquire_observation_lease(&self) -> Option<Rc<dyn Any>> {
707        let was_empty = !self.has_subscribers();
708        let lease = Rc::clone(&self.read_observation_lease);
709        if was_empty {
710            notify_subscriber_callbacks(&self.subscriber_callbacks);
711        }
712        Some(lease)
713    }
714
715    fn add_scope_observer(&self) -> bool {
716        let was_empty = !self.has_subscribers();
717        let count = self
718            .scope_observation_count
719            .get()
720            .checked_add(1)
721            .expect("state scope observation count overflow");
722        self.scope_observation_count.set(count);
723        was_empty
724    }
725
726    fn remove_scope_observers(&self, count: usize) {
727        if count == 0 {
728            return;
729        }
730        let remaining = self
731            .scope_observation_count
732            .get()
733            .checked_sub(count)
734            .expect("state scope observation count underflow");
735        self.scope_observation_count.set(remaining);
736    }
737
738    fn has_subscribers(&self) -> bool {
739        Rc::strong_count(&self.read_observation_lease) > 1 || self.scope_observation_count.get() > 0
740    }
741
742    fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
743        self.subscriber_callbacks
744            .borrow_mut()
745            .push(Rc::clone(&callback));
746        if notify {
747            callback();
748        }
749    }
750
751    fn notify_subscribers(&self) {
752        notify_subscriber_callbacks(&self.subscriber_callbacks);
753    }
754
755    fn notify_applied(&self) {
756        let observers = self.lock_apply_observers();
757        for observer in observers.iter() {
758            observer();
759        }
760    }
761
762    fn lock_weak_self(&self) -> MutexGuard<'_, Option<Weak<Self>>> {
763        self.weak_self
764            .lock()
765            .unwrap_or_else(PoisonError::into_inner)
766    }
767
768    fn lock_apply_observers(&self) -> MutexGuard<'_, Vec<Box<dyn Fn() + 'static>>> {
769        self.apply_observers
770            .lock()
771            .unwrap_or_else(PoisonError::into_inner)
772    }
773
774    fn upgrade_self(&self) -> Option<Arc<Self>> {
775        self.lock_weak_self().as_ref().and_then(Weak::upgrade)
776    }
777
778    #[inline]
779    pub(crate) fn id(&self) -> ObjectId {
780        self.id
781    }
782
783    pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
784        let record = self.readable_record_for_active_snapshot().ok()?;
785        record.try_with_value(f)
786    }
787
788    pub(crate) fn try_get(&self) -> Option<T> {
789        self.try_with_value(Clone::clone)
790    }
791
792    pub(crate) fn with_value<R>(&self, f: impl FnOnce(&T) -> R) -> R {
793        let record = self
794            .readable_record_for_active_snapshot()
795            .unwrap_or_else(|failure| panic!("{failure}"));
796        record.with_value(f)
797    }
798
799    pub(crate) fn get(&self) -> T {
800        self.with_value(Clone::clone)
801    }
802
803    pub(crate) fn set(&self, new_value: T) -> bool {
804        #[cfg(debug_assertions)]
805        {
806            let in_handler = crate::in_event_handler();
807            let in_snapshot = crate::in_applied_snapshot();
808            if in_handler && !in_snapshot {
809                log::warn!(
810                    target: "cranpose::state",
811                    "State modified in event handler without run_in_mutable_snapshot; \
812                     this can make updates invisible to other contexts. Wrap the handler \
813                     in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
814                    self.id
815                );
816            }
817        }
818
819        let snapshot = active_snapshot();
820        let snapshot_id = snapshot.snapshot_id();
821
822        match &snapshot {
823            AnySnapshot::Global(global) => {
824                let invalid = snapshot.invalid();
825                if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
826                    return false;
827                }
828
829                assert!(
830                    !global.has_pending_children(),
831                    "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
832                    global.pending_children(),
833                    self.id,
834                    snapshot_id
835                );
836
837                let mut written_state: Option<Arc<dyn StateObject>> = None;
838                if let Some(state) = self.upgrade_self() {
839                    let trait_object: Arc<dyn StateObject> = state;
840                    snapshot.record_write(trait_object.clone());
841                    written_state = Some(trait_object);
842                }
843                mark_update_write(self.id);
844
845                let new_id = allocate_record_id();
846                let record = new_overwritable_record_as_head_locked(self);
847                record.replace_value(new_value);
848                record.set_snapshot_id(new_id);
849                record.set_tombstone(false);
850                advance_global_snapshot(new_id);
851                self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
852
853                if !global.has_pending_children() {
854                    let mut cursor = record.next();
855                    while let Some(node) = cursor {
856                        if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
857                            node.clear_value();
858                            node.set_tombstone(true);
859                        }
860                        cursor = node.next();
861                    }
862                    self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
863                }
864
865                if let Some(modified) = written_state.as_ref() {
866                    crate::snapshot_v2::notify_apply_observers(
867                        std::slice::from_ref(modified),
868                        new_id,
869                    );
870                }
871            }
872            AnySnapshot::Mutable(_)
873            | AnySnapshot::NestedMutable(_)
874            | AnySnapshot::TransparentMutable(_) => {
875                let invalid = snapshot.invalid();
876                if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
877                    return false;
878                }
879
880                if let Some(state) = self.upgrade_self() {
881                    let trait_object: Arc<dyn StateObject> = state;
882                    snapshot.record_write(trait_object);
883                }
884                mark_update_write(self.id);
885
886                let record = self.writable_record(snapshot_id, &invalid);
887                record.replace_value(new_value);
888                self.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
889            }
890            AnySnapshot::Readonly(_)
891            | AnySnapshot::NestedReadonly(_)
892            | AnySnapshot::TransparentReadonly(_) => {
893                panic!("Cannot write to a read-only snapshot");
894            }
895        }
896
897        true
898    }
899}
900
901thread_local! {
902    static ACTIVE_UPDATES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
903    static PENDING_WRITES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
904}
905
906pub(crate) struct UpdateScope {
907    id: ObjectId,
908    finished: bool,
909}
910
911impl UpdateScope {
912    pub(crate) fn new(id: ObjectId) -> Self {
913        ACTIVE_UPDATES.with(|active| {
914            active.borrow_mut().insert(id);
915        });
916        PENDING_WRITES.with(|pending| {
917            pending.borrow_mut().remove(&id);
918        });
919        Self {
920            id,
921            finished: false,
922        }
923    }
924
925    pub(crate) fn finish(mut self) -> bool {
926        self.finished = true;
927        ACTIVE_UPDATES.with(|active| {
928            active.borrow_mut().remove(&self.id);
929        });
930        PENDING_WRITES.with(|pending| pending.borrow_mut().remove(&self.id))
931    }
932}
933
934impl Drop for UpdateScope {
935    fn drop(&mut self) {
936        if self.finished {
937            return;
938        }
939        ACTIVE_UPDATES.with(|active| {
940            active.borrow_mut().remove(&self.id);
941        });
942        PENDING_WRITES.with(|pending| {
943            pending.borrow_mut().remove(&self.id);
944        });
945    }
946}
947
948fn mark_update_write(id: ObjectId) {
949    ACTIVE_UPDATES.with(|active| {
950        if active.borrow().contains(&id) {
951            PENDING_WRITES.with(|pending| {
952                pending.borrow_mut().insert(id);
953            });
954        }
955    });
956}
957
958impl<T: Clone + 'static> SnapshotMutableState<T> {
959    fn try_readable_record(
960        &self,
961        snapshot_id: SnapshotId,
962        invalid: &SnapshotIdSet,
963    ) -> Option<Rc<StateRecord>> {
964        self.readable_for(snapshot_id, invalid)
965    }
966}
967
968impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
969    fn object_id(&self) -> ObjectId {
970        self.id
971    }
972
973    fn first_record(&self) -> Rc<StateRecord> {
974        self.head.clone_head()
975    }
976
977    fn try_readable_record(
978        &self,
979        snapshot_id: SnapshotId,
980        invalid: &SnapshotIdSet,
981    ) -> Option<Rc<StateRecord>> {
982        self.try_readable_record(snapshot_id, invalid)
983    }
984
985    fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
986        self.try_readable_record(snapshot_id, invalid)
987            .unwrap_or_else(|| {
988                panic!(
989                    "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
990                    self.id, snapshot_id
991                )
992            })
993    }
994
995    fn prepend_state_record(&self, record: Rc<StateRecord>) {
996        self.head.prepend(record);
997    }
998
999    fn observation_lease(&self) -> Option<Rc<dyn Any>> {
1000        self.acquire_observation_lease()
1001    }
1002
1003    fn merge_records(
1004        &self,
1005        previous: Rc<StateRecord>,
1006        current: Rc<StateRecord>,
1007        applied: Rc<StateRecord>,
1008    ) -> Option<Rc<StateRecord>> {
1009        let Some(current_value) = current.try_with_value(|value: &T| value.clone()) else {
1010            log::error!(
1011                "SnapshotMutableState::merge_records current record value missing or wrong type (state {:?}, current_id={})",
1012                self.id,
1013                current.snapshot_id()
1014            );
1015            return None;
1016        };
1017        let Some(applied_value) = applied.try_with_value(|value: &T| value.clone()) else {
1018            log::error!(
1019                "SnapshotMutableState::merge_records applied record value missing or wrong type (state {:?}, applied_id={})",
1020                self.id,
1021                applied.snapshot_id()
1022            );
1023            return None;
1024        };
1025        if self.policy.equivalent(&current_value, &applied_value) {
1026            return Some(current);
1027        }
1028
1029        let Some(previous_value) = previous.try_with_value(|value: &T| value.clone()) else {
1030            log::error!(
1031                "SnapshotMutableState::merge_records previous record value missing or wrong type (state {:?}, previous_id={})",
1032                self.id,
1033                previous.snapshot_id()
1034            );
1035            return None;
1036        };
1037        let merged = self
1038            .policy
1039            .merge(&previous_value, &current_value, &applied_value)?;
1040
1041        Some(StateRecord::new(applied.snapshot_id(), merged, None))
1042    }
1043
1044    fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1045        let head = self.first_record();
1046        let mut cursor = Some(head);
1047        while let Some(record) = cursor {
1048            if record.snapshot_id() == child_id {
1049                let Some(cloned) = record.try_with_value(|value: &T| value.clone()) else {
1050                    log::error!(
1051                        "SnapshotMutableState::promote_record child record value missing or wrong type (state {:?}, child_id={})",
1052                        self.id,
1053                        child_id
1054                    );
1055                    return Err("child record value missing or wrong type");
1056                };
1057                let new_id = allocate_record_id();
1058                let promoted = new_overwritable_record_as_head_locked(self);
1059                promoted.replace_value(cloned);
1060                promoted.set_tombstone(false);
1061                promoted.set_snapshot_id(new_id);
1062                advance_global_snapshot(new_id);
1063                self.notify_applied();
1064                self.assert_chain_integrity("promote_record", Some(child_id));
1065                return Ok(());
1066            }
1067            cursor = record.next();
1068        }
1069        log::error!(
1070            "SnapshotMutableState::promote_record missing child record (state {:?}, child_id={})",
1071            self.id,
1072            child_id
1073        );
1074        Err("missing child record")
1075    }
1076
1077    fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1078        let Some(value) = merged.try_with_value(|value: &T| value.clone()) else {
1079            log::error!(
1080                "SnapshotMutableState::commit_merged_record merged record value missing or wrong type (state {:?}, merged_id={})",
1081                self.id,
1082                merged.snapshot_id()
1083            );
1084            return Err("merged record value missing or wrong type");
1085        };
1086        let new_id = allocate_record_id();
1087        let committed = new_overwritable_record_as_head_locked(self);
1088        committed.replace_value(value);
1089        committed.set_tombstone(false);
1090        committed.set_snapshot_id(new_id);
1091        advance_global_snapshot(new_id);
1092        self.notify_applied();
1093        self.assert_chain_integrity("commit_merged_record", Some(new_id));
1094        Ok(new_id)
1095    }
1096
1097    fn overwrite_unused_records(&self) -> bool {
1098        overwrite_unused_records_locked::<T>(self)
1099    }
1100
1101    fn as_any(&self) -> &dyn Any {
1102        self
1103    }
1104}
1105
1106pub(crate) struct MutableStateInner<T: Clone + 'static> {
1107    pub(crate) state: Arc<SnapshotMutableState<T>>,
1108    pub(crate) watchers: RefCell<HashMap<ScopeId, RcWeak<RecomposeScopeInner>>>,
1109    runtime: RuntimeHandle,
1110    state_id: Cell<Option<StateId>>,
1111}
1112
1113fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<Rc<dyn Fn()>>>) {
1114    let registered = callbacks.borrow().len();
1115    for index in 0..registered {
1116        let callback = callbacks.borrow().get(index).map(Rc::clone);
1117        if let Some(callback) = callback {
1118            callback();
1119        }
1120    }
1121}
1122
1123fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, RcWeak<RecomposeScopeInner>>) {
1124    let len = watchers.len();
1125    let capacity = watchers.capacity();
1126    if capacity > len.saturating_mul(4).max(32) {
1127        watchers.shrink_to_fit();
1128    }
1129}
1130
1131impl<T: Clone + 'static> MutableStateInner<T> {
1132    pub(crate) fn new_with_policy(
1133        value: T,
1134        runtime: RuntimeHandle,
1135        policy: Arc<dyn MutationPolicy<T>>,
1136    ) -> Self {
1137        Self {
1138            state: SnapshotMutableState::new_in_arc(value, policy),
1139            watchers: RefCell::new(HashMap::default()),
1140            runtime,
1141            state_id: Cell::new(None),
1142        }
1143    }
1144
1145    pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1146        self.state_id.set(Some(state_id));
1147        let runtime_handle = self.runtime.clone();
1148        self.state.add_apply_observer(Box::new(move || {
1149            let runtime = runtime_handle.clone();
1150            runtime_handle.enqueue_ui_task(Box::new(move || {
1151                runtime.with_state_arena(|arena| {
1152                    let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1153                        inner.invalidate_watchers();
1154                    });
1155                });
1156            }));
1157        }));
1158    }
1159
1160    fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1161        let mut watchers = self.watchers.borrow_mut();
1162        let before = watchers.len();
1163        watchers.retain(|_, existing| existing.upgrade().is_some());
1164        self.state.remove_scope_observers(before - watchers.len());
1165        let registered = match watchers.get(&scope.id()) {
1166            Some(_) => false,
1167            _ => {
1168                watchers.insert(scope.id(), scope.downgrade());
1169                true
1170            }
1171        };
1172        drop(watchers);
1173        let became_subscribed = registered && self.state.add_scope_observer();
1174        (registered, became_subscribed)
1175    }
1176
1177    fn has_subscribers(&self) -> bool {
1178        let mut watchers = self.watchers.borrow_mut();
1179        let before = watchers.len();
1180        watchers.retain(|_, existing| existing.upgrade().is_some());
1181        self.state.remove_scope_observers(before - watchers.len());
1182        self.state.has_subscribers()
1183    }
1184
1185    pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1186        let mut watchers = self.watchers.borrow_mut();
1187        let removed = if watchers
1188            .get(&scope_id)
1189            .is_some_and(|weak| weak.upgrade().is_none())
1190        {
1191            watchers.remove(&scope_id);
1192            shrink_watchers_if_sparse(&mut watchers);
1193            true
1194        } else {
1195            false
1196        };
1197        drop(watchers);
1198        self.state.remove_scope_observers(usize::from(removed));
1199    }
1200
1201    fn state_id(&self) -> Option<StateId> {
1202        self.state_id.get()
1203    }
1204
1205    fn invalidate_watchers(&self) {
1206        let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1207            let mut watchers = self.watchers.borrow_mut();
1208            let before = watchers.len();
1209            let mut live = Vec::with_capacity(watchers.len());
1210            watchers.retain(|_, scope| {
1211                if let Some(inner) = scope.upgrade() {
1212                    live.push(RecomposeScope { inner });
1213                    true
1214                } else {
1215                    false
1216                }
1217            });
1218            let removed_count = before - watchers.len();
1219            shrink_watchers_if_sparse(&mut watchers);
1220            (live, removed_count)
1221        };
1222        self.state.remove_scope_observers(removed_count);
1223
1224        for watcher in watchers {
1225            debug_record_scope_invalidation::<T>(watcher.id(), self.state_id.get());
1226            if let Some(state_id) = self.state_id.get() {
1227                watcher.invalidate_from_state(state_id);
1228            } else {
1229                watcher.invalidate();
1230            }
1231        }
1232    }
1233}
1234
1235impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1236    fn drop(&mut self) {
1237        self.state
1238            .remove_scope_observers(self.watchers.get_mut().len());
1239    }
1240}
1241
1242fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1243    let Some(Some(scope)) =
1244        with_current_composer_opt(super::composer::Composer::current_state_invalidation_scope)
1245    else {
1246        return;
1247    };
1248    let (registered, became_subscribed) = inner.register_scope(&scope);
1249    if registered {
1250        if let Some(state_id) = inner.state_id() {
1251            scope.record_state_subscription(state_id);
1252        }
1253        if became_subscribed {
1254            inner.state.notify_subscribers();
1255        }
1256    }
1257}
1258
1259trait StateArenaHandle<T: Clone + 'static> {
1260    fn state_id(&self) -> StateId;
1261    fn runtime_id(&self) -> runtime::RuntimeId;
1262
1263    fn runtime_handle(&self) -> RuntimeHandle {
1264        runtime::runtime_handle_by_id(self.runtime_id())
1265            .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1266    }
1267
1268    fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1269        runtime::runtime_handle_by_id(self.runtime_id())
1270    }
1271
1272    fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1273        self.runtime_handle()
1274            .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1275    }
1276
1277    fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1278        self.runtime_handle_opt()?
1279            .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1280    }
1281}
1282
1283/// Cheap copyable read-only view of a state cell.
1284pub struct State<T: Clone + 'static> {
1285    id: StateId,
1286    runtime_id: runtime::RuntimeId,
1287    _marker: PhantomData<fn() -> T>,
1288}
1289
1290/// Cheap copyable mutable view of a state cell.
1291///
1292/// Ownership lives elsewhere: a composition slot, an [`OwnedMutableState`], or
1293/// the runtime for states created with [`crate::mutableStateOf`] /
1294/// [`MutableState::with_runtime`].
1295pub struct MutableState<T: Clone + 'static> {
1296    id: StateId,
1297    runtime_id: runtime::RuntimeId,
1298    _marker: PhantomData<fn() -> T>,
1299}
1300
1301/// Owning state handle for reclaimable state cells.
1302#[derive(Clone)]
1303pub struct OwnedMutableState<T: Clone + 'static> {
1304    state: MutableState<T>,
1305    _lease: Rc<runtime::StateHandleLease>,
1306    _marker: PhantomData<fn() -> T>,
1307}
1308
1309impl<T: Clone + 'static> PartialEq for State<T> {
1310    fn eq(&self, other: &Self) -> bool {
1311        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1312    }
1313}
1314
1315impl<T: Clone + 'static> Eq for State<T> {}
1316
1317impl<T: Clone + 'static> PartialEq for MutableState<T> {
1318    fn eq(&self, other: &Self) -> bool {
1319        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1320    }
1321}
1322
1323impl<T: Clone + 'static> Eq for MutableState<T> {}
1324
1325impl<T: Clone + 'static> Copy for State<T> {}
1326
1327impl<T: Clone + 'static> Clone for State<T> {
1328    fn clone(&self) -> Self {
1329        *self
1330    }
1331}
1332
1333impl<T: Clone + 'static> Copy for MutableState<T> {}
1334
1335impl<T: Clone + 'static> Clone for MutableState<T> {
1336    fn clone(&self) -> Self {
1337        *self
1338    }
1339}
1340
1341impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1342    fn state_id(&self) -> StateId {
1343        self.id
1344    }
1345
1346    fn runtime_id(&self) -> runtime::RuntimeId {
1347        self.runtime_id
1348    }
1349}
1350
1351impl<T: Clone + 'static> State<T> {
1352    fn subscribe_current_scope(&self) {
1353        self.with_inner(register_current_state_scope::<T>);
1354    }
1355
1356    pub fn is_alive(&self) -> bool {
1357        self.try_with_inner(|_| ()).is_some()
1358    }
1359
1360    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1361        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1362    }
1363
1364    pub fn try_value(&self) -> Option<T> {
1365        self.try_with_inner(|inner| inner.state.try_get())?
1366    }
1367
1368    /// Reads a copy of the value through `f` and subscribes the current
1369    /// scope. `f` may write this state; see [`Self::read`] for a read that
1370    /// borrows instead of copying.
1371    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1372        let value = self.with_inner(|inner| inner.state.get());
1373        self.subscribe_current_scope();
1374        f(&value)
1375    }
1376
1377    /// Reads the value in place through `f` and subscribes the current
1378    /// scope. Nothing is cloned, so a read of a large value costs nothing
1379    /// beyond `f`; in return `f` borrows the stored value and must not write
1380    /// this state.
1381    pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1382        let result = self.with_inner(|inner| inner.state.with_value(f));
1383        self.subscribe_current_scope();
1384        result
1385    }
1386
1387    pub fn value(&self) -> T {
1388        let value = self.with_inner(|inner| inner.state.get());
1389        self.subscribe_current_scope();
1390        value
1391    }
1392
1393    pub fn get(&self) -> T {
1394        self.value()
1395    }
1396
1397    pub fn has_subscribers(&self) -> bool {
1398        self.with_inner(MutableStateInner::has_subscribers)
1399    }
1400
1401    /// Runs `callback` each time this state gains its first subscriber, and
1402    /// right away when it already has one.
1403    ///
1404    /// The state owns `callback` for as long as the state lives, so nothing
1405    /// has to be kept alive on the caller's side.
1406    pub fn on_subscriber(&self, callback: impl Fn() + 'static) {
1407        let callback: Rc<dyn Fn()> = Rc::new(callback);
1408        self.with_inner(|inner| {
1409            inner
1410                .state
1411                .subscriber_callback(callback, inner.has_subscribers());
1412        });
1413    }
1414
1415    /// Counts this state as subscribed for as long as the returned guard
1416    /// lives, without observing it from any scope.
1417    ///
1418    /// Snapshot observers subscribe reads made under composition or the
1419    /// draw phase, and producers gated on [`State::has_subscribers`] — an
1420    /// infinite transition, for one — stop themselves when the last such
1421    /// subscriber leaves. A consumer that reads the value from a polling
1422    /// context (a frame effect stepping it into a derived state, a
1423    /// background task) is invisible to that accounting and would starve
1424    /// the producer it depends on. Holding this guard keeps the producer
1425    /// alive; dropping it releases the count.
1426    pub fn subscription_hold(&self) -> StateSubscriptionHold {
1427        StateSubscriptionHold {
1428            _lease: self
1429                .try_with_inner(|inner| inner.state.observation_lease())
1430                .flatten(),
1431        }
1432    }
1433}
1434
1435/// Keeps a [`State`] counted as subscribed while alive; see
1436/// [`State::subscription_hold`].
1437pub struct StateSubscriptionHold {
1438    _lease: Option<Rc<dyn Any>>,
1439}
1440
1441impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1442    fn state_id(&self) -> StateId {
1443        self.id
1444    }
1445
1446    fn runtime_id(&self) -> runtime::RuntimeId {
1447        self.runtime_id
1448    }
1449}
1450
1451impl<T: Clone + 'static> MutableState<T> {
1452    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1453        runtime.alloc_persistent_state(value)
1454    }
1455
1456    fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1457        Self {
1458            id,
1459            runtime_id,
1460            _marker: PhantomData,
1461        }
1462    }
1463
1464    pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1465        Self::from_parts(lease.id(), lease.runtime().id())
1466    }
1467
1468    pub fn is_alive(&self) -> bool {
1469        self.try_with_inner(|_| ()).is_some()
1470    }
1471
1472    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1473        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1474    }
1475
1476    pub fn try_value(&self) -> Option<T> {
1477        self.try_with_inner(|inner| inner.state.try_get())?
1478    }
1479
1480    pub fn as_state(&self) -> State<T> {
1481        State {
1482            id: self.id,
1483            runtime_id: self.runtime_id,
1484            _marker: PhantomData,
1485        }
1486    }
1487
1488    pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1489        let lease = self
1490            .runtime_handle_opt()?
1491            .retain_state_lease(self.state_id())?;
1492        Some(OwnedMutableState {
1493            state: *self,
1494            _lease: lease,
1495            _marker: PhantomData,
1496        })
1497    }
1498
1499    pub fn retain(&self) -> OwnedMutableState<T> {
1500        self.try_retain()
1501            .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1502    }
1503
1504    /// Reads a copy of the value through `f` and subscribes the current
1505    /// scope. `f` may write this state; see [`Self::read`] for a read that
1506    /// borrows instead of copying.
1507    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1508        let value = self.with_inner(|inner| inner.state.get());
1509        self.subscribe_current_scope();
1510        f(&value)
1511    }
1512
1513    /// Reads the value in place through `f` and subscribes the current
1514    /// scope. Nothing is cloned, so a read of a large value costs nothing
1515    /// beyond `f`; in return `f` borrows the stored value and must not write
1516    /// this state.
1517    pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1518        let result = self.with_inner(|inner| inner.state.with_value(f));
1519        self.subscribe_current_scope();
1520        result
1521    }
1522
1523    pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1524        let runtime = self.runtime_handle();
1525        runtime.assert_ui_thread();
1526        runtime.with_state_arena(|arena| {
1527            arena.with_typed::<T, R>(self.state_id(), |inner| {
1528                let mut value = inner.state.get();
1529                let tracker = UpdateScope::new(inner.state.id());
1530                let result = f(&mut value);
1531                let wrote_elsewhere = tracker.finish();
1532                if !wrote_elsewhere && inner.state.set(value) {
1533                    inner.invalidate_watchers();
1534                }
1535                result
1536            })
1537        })
1538    }
1539
1540    pub fn replace(&self, value: T) {
1541        let Some(runtime) = self.runtime_handle_opt() else {
1542            log::debug!(
1543                "MutableState::replace skipped: runtime {:?} dropped",
1544                self.runtime_id()
1545            );
1546            return;
1547        };
1548        runtime.assert_ui_thread();
1549        let replaced = runtime
1550            .try_with_state_arena(|arena| {
1551                arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1552                    if inner.state.set(value) {
1553                        inner.invalidate_watchers();
1554                    }
1555                })
1556            })
1557            .flatten();
1558        if replaced.is_none() {
1559            log::debug!(
1560                "MutableState::replace skipped: state cell released (slot={}, gen={})",
1561                self.state_id().slot(),
1562                self.state_id().generation(),
1563            );
1564        }
1565    }
1566
1567    pub fn set_value(&self, value: T) {
1568        self.replace(value);
1569    }
1570
1571    pub fn set(&self, value: T) {
1572        self.replace(value);
1573    }
1574
1575    pub fn value(&self) -> T {
1576        let value = self.with_inner(|inner| inner.state.get());
1577        self.subscribe_current_scope();
1578        value
1579    }
1580
1581    pub fn get(&self) -> T {
1582        self.value()
1583    }
1584
1585    pub fn get_non_reactive(&self) -> T {
1586        self.with_inner(|inner| inner.state.get())
1587    }
1588
1589    #[doc(hidden)]
1590    pub fn runtime_state_id(&self) -> StateId {
1591        self.state_id()
1592    }
1593
1594    #[doc(hidden)]
1595    pub fn subscribe_current_scope_only(&self) {
1596        self.subscribe_current_scope();
1597    }
1598
1599    fn subscribe_current_scope(&self) {
1600        self.with_inner(register_current_state_scope::<T>);
1601    }
1602
1603    #[cfg(test)]
1604    pub(crate) fn watcher_count(&self) -> usize {
1605        self.with_inner(|inner| inner.watchers.borrow().len())
1606    }
1607
1608    #[cfg(test)]
1609    pub(crate) fn watcher_capacity(&self) -> usize {
1610        self.with_inner(|inner| inner.watchers.borrow().capacity())
1611    }
1612
1613    #[cfg(test)]
1614    pub(crate) fn state_id_for_test(&self) -> StateId {
1615        self.state_id()
1616    }
1617
1618    #[cfg(test)]
1619    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1620        self.as_state().subscribe_scope_for_test(scope);
1621    }
1622}
1623
1624impl<T: Clone + 'static> OwnedMutableState<T> {
1625    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1626        let lease = runtime.alloc_state(value);
1627        Self {
1628            state: MutableState::from_lease(&lease),
1629            _lease: lease,
1630            _marker: PhantomData,
1631        }
1632    }
1633
1634    pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1635    where
1636        T: PartialEq,
1637    {
1638        Self::with_runtime_and_policy(value, runtime, Arc::new(StructuralEqual))
1639    }
1640
1641    pub(crate) fn with_runtime_and_policy(
1642        value: T,
1643        runtime: RuntimeHandle,
1644        policy: Arc<dyn MutationPolicy<T>>,
1645    ) -> Self {
1646        let lease = runtime.alloc_state_with_policy(value, policy);
1647        Self {
1648            state: MutableState::from_lease(&lease),
1649            _lease: lease,
1650            _marker: PhantomData,
1651        }
1652    }
1653
1654    pub fn handle(&self) -> MutableState<T> {
1655        self.state
1656    }
1657
1658    pub fn as_state(&self) -> State<T> {
1659        self.state.as_state()
1660    }
1661}
1662
1663impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1664    type Target = MutableState<T>;
1665
1666    fn deref(&self) -> &Self::Target {
1667        &self.state
1668    }
1669}
1670
1671#[cfg(test)]
1672impl<T: Clone + 'static> State<T> {
1673    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1674        self.with_inner(|inner| {
1675            let (registered, became_subscribed) = inner.register_scope(scope);
1676            if registered {
1677                if let Some(state_id) = inner.state_id() {
1678                    scope.record_state_subscription(state_id);
1679                }
1680                if became_subscribed {
1681                    inner.state.notify_subscribers();
1682                }
1683            }
1684        });
1685    }
1686}
1687
1688impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1689    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1690        if let Some(value) = self.try_value() {
1691            f.debug_struct("MutableState")
1692                .field("value", &value)
1693                .finish()
1694        } else {
1695            f.write_str("MutableState { value: <unavailable> }")
1696        }
1697    }
1698}
1699
1700#[derive(Clone)]
1701pub struct SnapshotStateList<T: Clone + 'static> {
1702    state: OwnedMutableState<Vec<T>>,
1703}
1704
1705impl<T: Clone + 'static> SnapshotStateList<T> {
1706    pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1707    where
1708        I: IntoIterator<Item = T>,
1709    {
1710        let initial: Vec<T> = values.into_iter().collect();
1711        Self {
1712            state: OwnedMutableState::with_runtime(initial, runtime),
1713        }
1714    }
1715
1716    pub fn as_state(&self) -> State<Vec<T>> {
1717        self.state.as_state()
1718    }
1719
1720    pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1721        self.state.handle()
1722    }
1723
1724    pub fn len(&self) -> usize {
1725        self.state.with(Vec::len)
1726    }
1727
1728    pub fn is_empty(&self) -> bool {
1729        self.len() == 0
1730    }
1731
1732    pub fn to_vec(&self) -> Vec<T> {
1733        self.state.with(Clone::clone)
1734    }
1735
1736    pub fn iter(&self) -> Vec<T> {
1737        self.to_vec()
1738    }
1739
1740    pub fn get(&self, index: usize) -> T {
1741        self.state.with(|values| values[index].clone())
1742    }
1743
1744    pub fn get_opt(&self, index: usize) -> Option<T> {
1745        self.state.with(|values| values.get(index).cloned())
1746    }
1747
1748    pub fn first(&self) -> Option<T> {
1749        self.get_opt(0)
1750    }
1751
1752    pub fn last(&self) -> Option<T> {
1753        self.state.with(|values| values.last().cloned())
1754    }
1755
1756    pub fn push(&self, value: T) {
1757        self.state.update(|values| values.push(value));
1758    }
1759
1760    pub fn extend<I>(&self, iter: I)
1761    where
1762        I: IntoIterator<Item = T>,
1763    {
1764        self.state.update(|values| values.extend(iter));
1765    }
1766
1767    pub fn insert(&self, index: usize, value: T) {
1768        self.state.update(|values| values.insert(index, value));
1769    }
1770
1771    pub fn set(&self, index: usize, value: T) -> T {
1772        self.state
1773            .update(|values| std::mem::replace(&mut values[index], value))
1774    }
1775
1776    pub fn remove(&self, index: usize) -> T {
1777        self.state.update(|values| values.remove(index))
1778    }
1779
1780    pub fn pop(&self) -> Option<T> {
1781        self.state.update(Vec::pop)
1782    }
1783
1784    pub fn clear(&self) {
1785        self.state.replace(Vec::new());
1786    }
1787
1788    pub fn retain<F>(&self, mut predicate: F)
1789    where
1790        F: FnMut(&T) -> bool,
1791    {
1792        self.state
1793            .update(|values| values.retain(|value| predicate(value)));
1794    }
1795
1796    pub fn replace_with<I>(&self, iter: I)
1797    where
1798        I: IntoIterator<Item = T>,
1799    {
1800        self.state.replace(iter.into_iter().collect());
1801    }
1802}
1803
1804impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
1805    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1806        let contents = self.to_vec();
1807        f.debug_struct("SnapshotStateList")
1808            .field("values", &contents)
1809            .finish()
1810    }
1811}
1812
1813#[derive(Clone)]
1814pub struct SnapshotStateMap<K, V>
1815where
1816    K: Clone + Eq + Hash + 'static,
1817    V: Clone + 'static,
1818{
1819    state: OwnedMutableState<HashMap<K, V>>,
1820}
1821
1822impl<K, V> SnapshotStateMap<K, V>
1823where
1824    K: Clone + Eq + Hash + 'static,
1825    V: Clone + 'static,
1826{
1827    pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
1828    where
1829        I: IntoIterator<Item = (K, V)>,
1830    {
1831        let map: HashMap<K, V> = pairs.into_iter().collect();
1832        Self {
1833            state: OwnedMutableState::with_runtime(map, runtime),
1834        }
1835    }
1836
1837    pub fn as_state(&self) -> State<HashMap<K, V>> {
1838        self.state.as_state()
1839    }
1840
1841    pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
1842        self.state.handle()
1843    }
1844
1845    pub fn len(&self) -> usize {
1846        self.state.with(std::collections::HashMap::len)
1847    }
1848
1849    pub fn is_empty(&self) -> bool {
1850        self.state.with(std::collections::HashMap::is_empty)
1851    }
1852
1853    pub fn contains_key(&self, key: &K) -> bool {
1854        self.state.with(|map| map.contains_key(key))
1855    }
1856
1857    pub fn get(&self, key: &K) -> Option<V> {
1858        self.state.with(|map| map.get(key).cloned())
1859    }
1860
1861    pub fn to_hash_map(&self) -> HashMap<K, V> {
1862        self.state.with(Clone::clone)
1863    }
1864
1865    pub fn insert(&self, key: K, value: V) -> Option<V> {
1866        self.state.update(|map| map.insert(key, value))
1867    }
1868
1869    pub fn extend<I>(&self, iter: I)
1870    where
1871        I: IntoIterator<Item = (K, V)>,
1872    {
1873        self.state.update(|map| map.extend(iter));
1874    }
1875
1876    pub fn remove(&self, key: &K) -> Option<V> {
1877        self.state.update(|map| map.remove(key))
1878    }
1879
1880    pub fn clear(&self) {
1881        self.state.replace(HashMap::default());
1882    }
1883
1884    pub fn retain<F>(&self, mut predicate: F)
1885    where
1886        F: FnMut(&K, &mut V) -> bool,
1887    {
1888        self.state.update(|map| map.retain(|k, v| predicate(k, v)));
1889    }
1890}
1891
1892impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
1893where
1894    K: Clone + Eq + Hash + fmt::Debug + 'static,
1895    V: Clone + fmt::Debug + 'static,
1896{
1897    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1898        let contents = self.to_hash_map();
1899        f.debug_struct("SnapshotStateMap")
1900            .field("entries", &contents)
1901            .finish()
1902    }
1903}
1904
1905pub(crate) struct DerivedState<T: Clone + 'static> {
1906    compute: Rc<dyn Fn() -> T>,
1907    pub(crate) state: OwnedMutableState<T>,
1908}
1909
1910impl<T: Clone + 'static> DerivedState<T> {
1911    pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
1912        let initial = compute();
1913        Self {
1914            compute,
1915            state: OwnedMutableState::with_runtime(initial, runtime),
1916        }
1917    }
1918
1919    pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
1920        self.compute = compute;
1921    }
1922
1923    pub(crate) fn recompute(&self) {
1924        let value = (self.compute)();
1925        self.state.set_value(value);
1926    }
1927}
1928
1929impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
1930    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1931        if let Some(value) = self.try_value() {
1932            f.debug_struct("State").field("value", &value).finish()
1933        } else {
1934            f.write_str("State { value: <unavailable> }")
1935        }
1936    }
1937}
1938
1939#[cfg(test)]
1940#[path = "tests/state_tests.rs"]
1941mod tests;