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    /// Adds `scope` to the state's watchers, saying whether it was new and
1161    /// whether the state thereby gained its first subscriber.
1162    ///
1163    /// Dead watchers still count as subscribers until pruned. They are
1164    /// pruned whenever no live one is left, so a reader arriving after every
1165    /// earlier one died still makes the state newly subscribed, and
1166    /// otherwise only when the map is full, before the insert would grow
1167    /// it: a state read by thousands of scopes, such as the density, must
1168    /// not scan them all for every new reader.
1169    fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1170        let mut watchers = self.watchers.borrow_mut();
1171        let id = scope.id();
1172        match watchers.get(&id) {
1173            Some(existing) if existing.strong_count() > 0 => return (false, false),
1174            Some(_) => {
1175                watchers.remove(&id);
1176                self.state.remove_scope_observers(1);
1177            }
1178            None => {}
1179        }
1180        let any_live = watchers.values().any(|watcher| watcher.strong_count() > 0);
1181        if !any_live || watchers.len() >= watchers.capacity() {
1182            let before = watchers.len();
1183            watchers.retain(|_, watcher| watcher.strong_count() > 0);
1184            self.state.remove_scope_observers(before - watchers.len());
1185        }
1186        watchers.insert(id, scope.downgrade());
1187        drop(watchers);
1188        (true, self.state.add_scope_observer())
1189    }
1190
1191    fn has_subscribers(&self) -> bool {
1192        let mut watchers = self.watchers.borrow_mut();
1193        let before = watchers.len();
1194        watchers.retain(|_, existing| existing.upgrade().is_some());
1195        self.state.remove_scope_observers(before - watchers.len());
1196        self.state.has_subscribers()
1197    }
1198
1199    pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1200        let mut watchers = self.watchers.borrow_mut();
1201        let removed = if watchers
1202            .get(&scope_id)
1203            .is_some_and(|weak| weak.upgrade().is_none())
1204        {
1205            watchers.remove(&scope_id);
1206            shrink_watchers_if_sparse(&mut watchers);
1207            true
1208        } else {
1209            false
1210        };
1211        drop(watchers);
1212        self.state.remove_scope_observers(usize::from(removed));
1213    }
1214
1215    fn state_id(&self) -> Option<StateId> {
1216        self.state_id.get()
1217    }
1218
1219    fn invalidate_watchers(&self) {
1220        let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1221            let mut watchers = self.watchers.borrow_mut();
1222            let before = watchers.len();
1223            let mut live = Vec::with_capacity(watchers.len());
1224            watchers.retain(|_, scope| {
1225                if let Some(inner) = scope.upgrade() {
1226                    live.push(RecomposeScope { inner });
1227                    true
1228                } else {
1229                    false
1230                }
1231            });
1232            let removed_count = before - watchers.len();
1233            shrink_watchers_if_sparse(&mut watchers);
1234            (live, removed_count)
1235        };
1236        self.state.remove_scope_observers(removed_count);
1237
1238        for watcher in watchers {
1239            debug_record_scope_invalidation(
1240                watcher.id(),
1241                self.state_id.get(),
1242                std::any::type_name::<T>(),
1243            );
1244            if let Some(state_id) = self.state_id.get() {
1245                watcher.invalidate_from_state(state_id);
1246            } else {
1247                watcher.invalidate();
1248            }
1249        }
1250    }
1251}
1252
1253impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1254    fn drop(&mut self) {
1255        self.state
1256            .remove_scope_observers(self.watchers.get_mut().len());
1257    }
1258}
1259
1260fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1261    let Some(Some(scope)) =
1262        with_current_composer_opt(super::composer::Composer::current_state_invalidation_scope)
1263    else {
1264        return;
1265    };
1266    let (registered, became_subscribed) = inner.register_scope(&scope);
1267    if registered {
1268        if let Some(state_id) = inner.state_id() {
1269            scope.record_state_subscription(state_id);
1270        }
1271        if became_subscribed {
1272            inner.state.notify_subscribers();
1273        }
1274    }
1275}
1276
1277trait StateArenaHandle<T: Clone + 'static> {
1278    fn state_id(&self) -> StateId;
1279    fn runtime_id(&self) -> runtime::RuntimeId;
1280
1281    fn runtime_handle(&self) -> RuntimeHandle {
1282        runtime::runtime_handle_by_id(self.runtime_id())
1283            .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1284    }
1285
1286    fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1287        runtime::runtime_handle_by_id(self.runtime_id())
1288    }
1289
1290    fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1291        self.runtime_handle()
1292            .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1293    }
1294
1295    fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1296        self.runtime_handle_opt()?
1297            .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1298    }
1299}
1300
1301/// Cheap copyable read-only view of a state cell.
1302pub struct State<T: Clone + 'static> {
1303    id: StateId,
1304    runtime_id: runtime::RuntimeId,
1305    _marker: PhantomData<fn() -> T>,
1306}
1307
1308/// Cheap copyable mutable view of a state cell.
1309///
1310/// Ownership lives elsewhere: a composition slot, an [`OwnedMutableState`], or
1311/// the runtime for states created with [`crate::mutableStateOf`] /
1312/// [`MutableState::with_runtime`].
1313pub struct MutableState<T: Clone + 'static> {
1314    id: StateId,
1315    runtime_id: runtime::RuntimeId,
1316    _marker: PhantomData<fn() -> T>,
1317}
1318
1319/// Owning state handle for reclaimable state cells.
1320#[derive(Clone)]
1321pub struct OwnedMutableState<T: Clone + 'static> {
1322    state: MutableState<T>,
1323    _lease: Rc<runtime::StateHandleLease>,
1324    _marker: PhantomData<fn() -> T>,
1325}
1326
1327impl<T: Clone + 'static> PartialEq for State<T> {
1328    fn eq(&self, other: &Self) -> bool {
1329        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1330    }
1331}
1332
1333impl<T: Clone + 'static> Eq for State<T> {}
1334
1335impl<T: Clone + 'static> PartialEq for MutableState<T> {
1336    fn eq(&self, other: &Self) -> bool {
1337        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1338    }
1339}
1340
1341impl<T: Clone + 'static> Eq for MutableState<T> {}
1342
1343impl<T: Clone + 'static> Copy for State<T> {}
1344
1345impl<T: Clone + 'static> Clone for State<T> {
1346    fn clone(&self) -> Self {
1347        *self
1348    }
1349}
1350
1351impl<T: Clone + 'static> Copy for MutableState<T> {}
1352
1353impl<T: Clone + 'static> Clone for MutableState<T> {
1354    fn clone(&self) -> Self {
1355        *self
1356    }
1357}
1358
1359impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1360    fn state_id(&self) -> StateId {
1361        self.id
1362    }
1363
1364    fn runtime_id(&self) -> runtime::RuntimeId {
1365        self.runtime_id
1366    }
1367}
1368
1369impl<T: Clone + 'static> State<T> {
1370    fn subscribe_current_scope(&self) {
1371        self.with_inner(register_current_state_scope::<T>);
1372    }
1373
1374    pub fn is_alive(&self) -> bool {
1375        self.try_with_inner(|_| ()).is_some()
1376    }
1377
1378    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1379        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1380    }
1381
1382    pub fn try_value(&self) -> Option<T> {
1383        self.try_with_inner(|inner| inner.state.try_get())?
1384    }
1385
1386    /// Reads a copy of the value through `f` and subscribes the current
1387    /// scope. `f` may write this state; see [`Self::read`] for a read that
1388    /// borrows instead of copying.
1389    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1390        let value = self.with_inner(|inner| inner.state.get());
1391        self.subscribe_current_scope();
1392        f(&value)
1393    }
1394
1395    /// Reads the value in place through `f` and subscribes the current
1396    /// scope. Nothing is cloned, so a read of a large value costs nothing
1397    /// beyond `f`; in return `f` borrows the stored value and must not write
1398    /// this state.
1399    pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1400        let result = self.with_inner(|inner| inner.state.with_value(f));
1401        self.subscribe_current_scope();
1402        result
1403    }
1404
1405    pub fn value(&self) -> T {
1406        let value = self.with_inner(|inner| inner.state.get());
1407        self.subscribe_current_scope();
1408        value
1409    }
1410
1411    pub fn get(&self) -> T {
1412        self.value()
1413    }
1414
1415    pub fn has_subscribers(&self) -> bool {
1416        self.with_inner(MutableStateInner::has_subscribers)
1417    }
1418
1419    /// Runs `callback` each time this state gains its first subscriber, and
1420    /// right away when it already has one.
1421    ///
1422    /// The state owns `callback` for as long as the state lives, so nothing
1423    /// has to be kept alive on the caller's side.
1424    pub fn on_subscriber(&self, callback: impl Fn() + 'static) {
1425        let callback: Rc<dyn Fn()> = Rc::new(callback);
1426        self.with_inner(|inner| {
1427            inner
1428                .state
1429                .subscriber_callback(callback, inner.has_subscribers());
1430        });
1431    }
1432
1433    /// Counts this state as subscribed for as long as the returned guard
1434    /// lives, without observing it from any scope.
1435    ///
1436    /// Snapshot observers subscribe reads made under composition or the
1437    /// draw phase, and producers gated on [`State::has_subscribers`] — an
1438    /// infinite transition, for one — stop themselves when the last such
1439    /// subscriber leaves. A consumer that reads the value from a polling
1440    /// context (a frame effect stepping it into a derived state, a
1441    /// background task) is invisible to that accounting and would starve
1442    /// the producer it depends on. Holding this guard keeps the producer
1443    /// alive; dropping it releases the count.
1444    pub fn subscription_hold(&self) -> StateSubscriptionHold {
1445        StateSubscriptionHold {
1446            _lease: self
1447                .try_with_inner(|inner| inner.state.observation_lease())
1448                .flatten(),
1449        }
1450    }
1451}
1452
1453/// Keeps a [`State`] counted as subscribed while alive; see
1454/// [`State::subscription_hold`].
1455pub struct StateSubscriptionHold {
1456    _lease: Option<Rc<dyn Any>>,
1457}
1458
1459impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1460    fn state_id(&self) -> StateId {
1461        self.id
1462    }
1463
1464    fn runtime_id(&self) -> runtime::RuntimeId {
1465        self.runtime_id
1466    }
1467}
1468
1469impl<T: Clone + 'static> MutableState<T> {
1470    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1471        runtime.alloc_persistent_state(value)
1472    }
1473
1474    fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1475        Self {
1476            id,
1477            runtime_id,
1478            _marker: PhantomData,
1479        }
1480    }
1481
1482    pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1483        Self::from_parts(lease.id(), lease.runtime().id())
1484    }
1485
1486    pub fn is_alive(&self) -> bool {
1487        self.try_with_inner(|_| ()).is_some()
1488    }
1489
1490    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1491        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1492    }
1493
1494    pub fn try_value(&self) -> Option<T> {
1495        self.try_with_inner(|inner| inner.state.try_get())?
1496    }
1497
1498    pub fn as_state(&self) -> State<T> {
1499        State {
1500            id: self.id,
1501            runtime_id: self.runtime_id,
1502            _marker: PhantomData,
1503        }
1504    }
1505
1506    pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1507        let lease = self
1508            .runtime_handle_opt()?
1509            .retain_state_lease(self.state_id())?;
1510        Some(OwnedMutableState {
1511            state: *self,
1512            _lease: lease,
1513            _marker: PhantomData,
1514        })
1515    }
1516
1517    pub fn retain(&self) -> OwnedMutableState<T> {
1518        self.try_retain()
1519            .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1520    }
1521
1522    /// Reads a copy of the value through `f` and subscribes the current
1523    /// scope. `f` may write this state; see [`Self::read`] for a read that
1524    /// borrows instead of copying.
1525    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1526        let value = self.with_inner(|inner| inner.state.get());
1527        self.subscribe_current_scope();
1528        f(&value)
1529    }
1530
1531    /// Reads the value in place through `f` and subscribes the current
1532    /// scope. Nothing is cloned, so a read of a large value costs nothing
1533    /// beyond `f`; in return `f` borrows the stored value and must not write
1534    /// this state.
1535    pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1536        let result = self.with_inner(|inner| inner.state.with_value(f));
1537        self.subscribe_current_scope();
1538        result
1539    }
1540
1541    pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1542        let runtime = self.runtime_handle();
1543        runtime.assert_ui_thread();
1544        runtime.with_state_arena(|arena| {
1545            arena.with_typed::<T, R>(self.state_id(), |inner| {
1546                let mut value = inner.state.get();
1547                let tracker = UpdateScope::new(inner.state.id());
1548                let result = f(&mut value);
1549                let wrote_elsewhere = tracker.finish();
1550                if !wrote_elsewhere && inner.state.set(value) {
1551                    inner.invalidate_watchers();
1552                }
1553                result
1554            })
1555        })
1556    }
1557
1558    pub fn replace(&self, value: T) {
1559        let Some(runtime) = self.runtime_handle_opt() else {
1560            log::debug!(
1561                "MutableState::replace skipped: runtime {:?} dropped",
1562                self.runtime_id()
1563            );
1564            return;
1565        };
1566        runtime.assert_ui_thread();
1567        let replaced = runtime
1568            .try_with_state_arena(|arena| {
1569                arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1570                    if inner.state.set(value) {
1571                        inner.invalidate_watchers();
1572                    }
1573                })
1574            })
1575            .flatten();
1576        if replaced.is_none() {
1577            log::debug!(
1578                "MutableState::replace skipped: state cell released (slot={}, gen={})",
1579                self.state_id().slot(),
1580                self.state_id().generation(),
1581            );
1582        }
1583    }
1584
1585    pub fn set_value(&self, value: T) {
1586        self.replace(value);
1587    }
1588
1589    pub fn set(&self, value: T) {
1590        self.replace(value);
1591    }
1592
1593    pub fn value(&self) -> T {
1594        let value = self.with_inner(|inner| inner.state.get());
1595        self.subscribe_current_scope();
1596        value
1597    }
1598
1599    pub fn get(&self) -> T {
1600        self.value()
1601    }
1602
1603    pub fn get_non_reactive(&self) -> T {
1604        self.with_inner(|inner| inner.state.get())
1605    }
1606
1607    #[doc(hidden)]
1608    pub fn runtime_state_id(&self) -> StateId {
1609        self.state_id()
1610    }
1611
1612    #[doc(hidden)]
1613    pub fn subscribe_current_scope_only(&self) {
1614        self.subscribe_current_scope();
1615    }
1616
1617    fn subscribe_current_scope(&self) {
1618        self.with_inner(register_current_state_scope::<T>);
1619    }
1620
1621    #[cfg(test)]
1622    pub(crate) fn watcher_count(&self) -> usize {
1623        self.with_inner(|inner| inner.watchers.borrow().len())
1624    }
1625
1626    #[cfg(test)]
1627    pub(crate) fn watcher_capacity(&self) -> usize {
1628        self.with_inner(|inner| inner.watchers.borrow().capacity())
1629    }
1630
1631    #[cfg(test)]
1632    pub(crate) fn state_id_for_test(&self) -> StateId {
1633        self.state_id()
1634    }
1635
1636    #[cfg(test)]
1637    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1638        self.as_state().subscribe_scope_for_test(scope);
1639    }
1640}
1641
1642impl<T: Clone + 'static> OwnedMutableState<T> {
1643    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1644        let lease = runtime.alloc_state(value);
1645        Self {
1646            state: MutableState::from_lease(&lease),
1647            _lease: lease,
1648            _marker: PhantomData,
1649        }
1650    }
1651
1652    pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1653    where
1654        T: PartialEq,
1655    {
1656        Self::with_runtime_and_policy(value, runtime, Arc::new(StructuralEqual))
1657    }
1658
1659    pub(crate) fn with_runtime_and_policy(
1660        value: T,
1661        runtime: RuntimeHandle,
1662        policy: Arc<dyn MutationPolicy<T>>,
1663    ) -> Self {
1664        let lease = runtime.alloc_state_with_policy(value, policy);
1665        Self {
1666            state: MutableState::from_lease(&lease),
1667            _lease: lease,
1668            _marker: PhantomData,
1669        }
1670    }
1671
1672    pub fn handle(&self) -> MutableState<T> {
1673        self.state
1674    }
1675
1676    pub fn as_state(&self) -> State<T> {
1677        self.state.as_state()
1678    }
1679}
1680
1681impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1682    type Target = MutableState<T>;
1683
1684    fn deref(&self) -> &Self::Target {
1685        &self.state
1686    }
1687}
1688
1689#[cfg(test)]
1690impl<T: Clone + 'static> State<T> {
1691    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1692        self.with_inner(|inner| {
1693            let (registered, became_subscribed) = inner.register_scope(scope);
1694            if registered {
1695                if let Some(state_id) = inner.state_id() {
1696                    scope.record_state_subscription(state_id);
1697                }
1698                if became_subscribed {
1699                    inner.state.notify_subscribers();
1700                }
1701            }
1702        });
1703    }
1704}
1705
1706impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1707    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1708        if let Some(value) = self.try_value() {
1709            f.debug_struct("MutableState")
1710                .field("value", &value)
1711                .finish()
1712        } else {
1713            f.write_str("MutableState { value: <unavailable> }")
1714        }
1715    }
1716}
1717
1718#[derive(Clone)]
1719pub struct SnapshotStateList<T: Clone + 'static> {
1720    state: OwnedMutableState<Vec<T>>,
1721}
1722
1723impl<T: Clone + 'static> SnapshotStateList<T> {
1724    pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1725    where
1726        I: IntoIterator<Item = T>,
1727    {
1728        let initial: Vec<T> = values.into_iter().collect();
1729        Self {
1730            state: OwnedMutableState::with_runtime(initial, runtime),
1731        }
1732    }
1733
1734    pub fn as_state(&self) -> State<Vec<T>> {
1735        self.state.as_state()
1736    }
1737
1738    pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1739        self.state.handle()
1740    }
1741
1742    pub fn len(&self) -> usize {
1743        self.state.with(Vec::len)
1744    }
1745
1746    pub fn is_empty(&self) -> bool {
1747        self.len() == 0
1748    }
1749
1750    pub fn to_vec(&self) -> Vec<T> {
1751        self.state.with(Clone::clone)
1752    }
1753
1754    pub fn iter(&self) -> Vec<T> {
1755        self.to_vec()
1756    }
1757
1758    pub fn get(&self, index: usize) -> T {
1759        self.state.with(|values| values[index].clone())
1760    }
1761
1762    pub fn get_opt(&self, index: usize) -> Option<T> {
1763        self.state.with(|values| values.get(index).cloned())
1764    }
1765
1766    pub fn first(&self) -> Option<T> {
1767        self.get_opt(0)
1768    }
1769
1770    pub fn last(&self) -> Option<T> {
1771        self.state.with(|values| values.last().cloned())
1772    }
1773
1774    pub fn push(&self, value: T) {
1775        self.state.update(|values| values.push(value));
1776    }
1777
1778    pub fn extend<I>(&self, iter: I)
1779    where
1780        I: IntoIterator<Item = T>,
1781    {
1782        self.state.update(|values| values.extend(iter));
1783    }
1784
1785    pub fn insert(&self, index: usize, value: T) {
1786        self.state.update(|values| values.insert(index, value));
1787    }
1788
1789    pub fn set(&self, index: usize, value: T) -> T {
1790        self.state
1791            .update(|values| std::mem::replace(&mut values[index], value))
1792    }
1793
1794    pub fn remove(&self, index: usize) -> T {
1795        self.state.update(|values| values.remove(index))
1796    }
1797
1798    pub fn pop(&self) -> Option<T> {
1799        self.state.update(Vec::pop)
1800    }
1801
1802    pub fn clear(&self) {
1803        self.state.replace(Vec::new());
1804    }
1805
1806    pub fn retain<F>(&self, mut predicate: F)
1807    where
1808        F: FnMut(&T) -> bool,
1809    {
1810        self.state
1811            .update(|values| values.retain(|value| predicate(value)));
1812    }
1813
1814    pub fn replace_with<I>(&self, iter: I)
1815    where
1816        I: IntoIterator<Item = T>,
1817    {
1818        self.state.replace(iter.into_iter().collect());
1819    }
1820}
1821
1822impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
1823    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1824        let contents = self.to_vec();
1825        f.debug_struct("SnapshotStateList")
1826            .field("values", &contents)
1827            .finish()
1828    }
1829}
1830
1831#[derive(Clone)]
1832pub struct SnapshotStateMap<K, V>
1833where
1834    K: Clone + Eq + Hash + 'static,
1835    V: Clone + 'static,
1836{
1837    state: OwnedMutableState<HashMap<K, V>>,
1838}
1839
1840impl<K, V> SnapshotStateMap<K, V>
1841where
1842    K: Clone + Eq + Hash + 'static,
1843    V: Clone + 'static,
1844{
1845    pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
1846    where
1847        I: IntoIterator<Item = (K, V)>,
1848    {
1849        let map: HashMap<K, V> = pairs.into_iter().collect();
1850        Self {
1851            state: OwnedMutableState::with_runtime(map, runtime),
1852        }
1853    }
1854
1855    pub fn as_state(&self) -> State<HashMap<K, V>> {
1856        self.state.as_state()
1857    }
1858
1859    pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
1860        self.state.handle()
1861    }
1862
1863    pub fn len(&self) -> usize {
1864        self.state.with(std::collections::HashMap::len)
1865    }
1866
1867    pub fn is_empty(&self) -> bool {
1868        self.state.with(std::collections::HashMap::is_empty)
1869    }
1870
1871    pub fn contains_key(&self, key: &K) -> bool {
1872        self.state.with(|map| map.contains_key(key))
1873    }
1874
1875    pub fn get(&self, key: &K) -> Option<V> {
1876        self.state.with(|map| map.get(key).cloned())
1877    }
1878
1879    pub fn to_hash_map(&self) -> HashMap<K, V> {
1880        self.state.with(Clone::clone)
1881    }
1882
1883    pub fn insert(&self, key: K, value: V) -> Option<V> {
1884        self.state.update(|map| map.insert(key, value))
1885    }
1886
1887    pub fn extend<I>(&self, iter: I)
1888    where
1889        I: IntoIterator<Item = (K, V)>,
1890    {
1891        self.state.update(|map| map.extend(iter));
1892    }
1893
1894    pub fn remove(&self, key: &K) -> Option<V> {
1895        self.state.update(|map| map.remove(key))
1896    }
1897
1898    pub fn clear(&self) {
1899        self.state.replace(HashMap::default());
1900    }
1901
1902    pub fn retain<F>(&self, mut predicate: F)
1903    where
1904        F: FnMut(&K, &mut V) -> bool,
1905    {
1906        self.state.update(|map| map.retain(|k, v| predicate(k, v)));
1907    }
1908}
1909
1910impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
1911where
1912    K: Clone + Eq + Hash + fmt::Debug + 'static,
1913    V: Clone + fmt::Debug + 'static,
1914{
1915    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1916        let contents = self.to_hash_map();
1917        f.debug_struct("SnapshotStateMap")
1918            .field("entries", &contents)
1919            .finish()
1920    }
1921}
1922
1923pub(crate) struct DerivedState<T: Clone + PartialEq + 'static> {
1924    compute: Rc<dyn Fn() -> T>,
1925    pub(crate) state: OwnedMutableState<T>,
1926}
1927
1928impl<T: Clone + PartialEq + 'static> DerivedState<T> {
1929    pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
1930        let initial = compute();
1931        Self {
1932            compute,
1933            state: OwnedMutableState::with_runtime_structural_eq(initial, runtime),
1934        }
1935    }
1936
1937    pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
1938        self.compute = compute;
1939    }
1940
1941    pub(crate) fn recompute(&self) {
1942        let value = (self.compute)();
1943        self.state.set_value(value);
1944    }
1945}
1946
1947impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
1948    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1949        if let Some(value) = self.try_value() {
1950            f.debug_struct("State").field("value", &value).finish()
1951        } else {
1952            f.write_str("State { value: <unavailable> }")
1953        }
1954    }
1955}
1956
1957#[cfg(test)]
1958#[path = "tests/state_tests.rs"]
1959mod tests;