Skip to main content

cranpose_core/
state.rs

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