Skip to main content

cranpose_core/
state.rs

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