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
502std::thread_local! {
503    // Record values displaced inside snapshot bookkeeping, and how many
504    // operations that hold them back are running.
505    static RETIRED_RECORD_VALUES: RefCell<Vec<Box<dyn RecordValue>>> =
506        const { RefCell::new(Vec::new()) };
507    static RETIRE_DEPTH: Cell<usize> = const { Cell::new(0) };
508}
509
510/// Drops `value`, a record value displaced inside snapshot bookkeeping, once
511/// the outermost [`RetiredValuesScope`] ends, or now when none is open: its
512/// destructor may read or write state.
513pub(crate) fn retire_record_value(value: Box<dyn RecordValue>) {
514    if RETIRE_DEPTH.with(Cell::get) == 0 {
515        drop(value);
516    } else {
517        RETIRED_RECORD_VALUES.with(|retired| retired.borrow_mut().push(value));
518    }
519}
520
521/// Holds back the drop of retired record values until the outermost scope
522/// ends, so no destructor runs while snapshot bookkeeping is half done.
523pub(crate) struct RetiredValuesScope(());
524
525impl RetiredValuesScope {
526    pub(crate) fn enter() -> Self {
527        RETIRE_DEPTH.with(|depth| depth.set(depth.get() + 1));
528        Self(())
529    }
530}
531
532impl Drop for RetiredValuesScope {
533    fn drop(&mut self) {
534        let outermost = RETIRE_DEPTH.with(|depth| {
535            depth.set(depth.get() - 1);
536            depth.get() == 0
537        });
538        if outermost {
539            let retired =
540                RETIRED_RECORD_VALUES.with(|retired| std::mem::take(&mut *retired.borrow_mut()));
541            drop(retired);
542        }
543    }
544}
545
546/// Marks the records of `state` that no open snapshot can read as free for
547/// reuse, retiring their values, and reports whether the state still keeps
548/// more than one record.
549pub(crate) fn overwrite_unused_records_locked(state: &dyn StateObject) -> bool {
550    let mut current = Some(state.first_record());
551    let mut valid_record: Option<Rc<StateRecord>> = None;
552
553    let reuse_limit =
554        lowest_pinned_snapshot().unwrap_or_else(crate::snapshot_v2::peek_next_snapshot_id);
555
556    let mut retained_records = 0;
557
558    while let Some(record) = current {
559        let current_id = record.snapshot_id();
560
561        if current_id == INVALID_SNAPSHOT_ID {
562        } else if current_id < reuse_limit {
563            match valid_record.as_ref() {
564                None => {
565                    valid_record = Some(Rc::clone(&record));
566                    retained_records += 1;
567                }
568                Some(valid) => {
569                    let record_to_overwrite = if current_id < valid.snapshot_id() {
570                        Rc::clone(&record)
571                    } else {
572                        let to_overwrite = Rc::clone(valid);
573                        valid_record = Some(Rc::clone(&record));
574                        to_overwrite
575                    };
576                    record_to_overwrite.set_snapshot_id(INVALID_SNAPSHOT_ID);
577                    if let Some(value) = record_to_overwrite.clear_value() {
578                        retire_record_value(value);
579                    }
580                }
581            }
582        } else {
583            retained_records += 1;
584        }
585
586        current = record.next();
587    }
588
589    retained_records > 1
590}
591
592fn active_snapshot() -> AnySnapshot {
593    current_snapshot().unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()))
594}
595
596pub(crate) trait MutationPolicy<T>: Send + Sync {
597    fn equivalent(&self, a: &T, b: &T) -> bool;
598    fn merge(&self, _previous: &T, _current: &T, _applied: &T) -> Option<T> {
599        None
600    }
601}
602
603pub(crate) struct NeverEqual;
604
605impl<T> MutationPolicy<T> for NeverEqual {
606    fn equivalent(&self, _a: &T, _b: &T) -> bool {
607        false
608    }
609}
610
611pub(crate) struct StructuralEqual;
612
613impl<T: PartialEq> MutationPolicy<T> for StructuralEqual {
614    fn equivalent(&self, a: &T, b: &T) -> bool {
615        a == b
616    }
617}
618
619pub trait StateObject: Any {
620    fn object_id(&self) -> ObjectId;
621    fn first_record(&self) -> Rc<StateRecord>;
622    fn try_readable_record(
623        &self,
624        snapshot_id: SnapshotId,
625        invalid: &SnapshotIdSet,
626    ) -> Option<Rc<StateRecord>>;
627    fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord>;
628
629    fn prepend_state_record(&self, record: Rc<StateRecord>);
630
631    fn merge_records(
632        &self,
633        _previous: Rc<StateRecord>,
634        _current: Rc<StateRecord>,
635        _applied: Rc<StateRecord>,
636    ) -> Option<Rc<StateRecord>> {
637        None
638    }
639
640    fn commit_merged_record(&self, _merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
641        Err("StateObject does not support merged record commits")
642    }
643    fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str>;
644
645    fn overwrite_unused_records(&self) -> bool {
646        false
647    }
648
649    fn observation_lease(&self) -> Option<Rc<dyn Any>> {
650        None
651    }
652
653    fn as_any(&self) -> &dyn Any;
654}
655
656pub(crate) struct SnapshotMutableState<T> {
657    head: CurrentRecord,
658    policy: Rc<dyn MutationPolicy<T>>,
659    id: ObjectId,
660    weak_self: Weak<Self>,
661    apply_observer: ApplyObserver,
662    read_observation_lease: Rc<()>,
663    scope_observation_count: Cell<usize>,
664    subscriber_callbacks: RefCell<Vec<Rc<dyn Fn()>>>,
665}
666
667#[inline(always)]
668fn assert_record_chain(
669    head: &CurrentRecord,
670    id: ObjectId,
671    caller: &str,
672    snapshot_context: Option<SnapshotId>,
673) {
674    if should_check_chain_integrity() {
675        check_chain_integrity(&head.clone_head(), id, caller, snapshot_context);
676    }
677}
678
679enum Merge<'a, T> {
680    KeepCurrent,
681    Value(Option<T>),
682    Missing(&'static str, &'a Rc<StateRecord>),
683}
684
685type ApplyObserver = OnceCell<Box<dyn Fn() + 'static>>;
686
687fn notify_applied(observer: &ApplyObserver) {
688    if let Some(observer) = observer.get() {
689        observer();
690    }
691}
692
693fn finish_apply(
694    observer: &ApplyObserver,
695    head: &CurrentRecord,
696    id: ObjectId,
697    caller: &str,
698    snapshot_id: SnapshotId,
699) {
700    notify_applied(observer);
701    assert_record_chain(head, id, caller, Some(snapshot_id));
702}
703
704// What a write does that does not depend on the state's value type, so it
705// is compiled once instead of once per state type.
706struct WriteTarget<'a> {
707    head: &'a CurrentRecord,
708    id: ObjectId,
709    state: &'a dyn StateObject,
710}
711
712struct GlobalWrite {
713    record: Rc<StateRecord>,
714    new_id: SnapshotId,
715    written: Option<Rc<dyn StateObject>>,
716}
717
718impl WriteTarget<'_> {
719    fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
720        assert_record_chain(self.head, self.id, caller, snapshot_context);
721    }
722
723    fn begin_global_write(
724        &self,
725        snapshot: &AnySnapshot,
726        global: &GlobalSnapshot,
727        snapshot_id: SnapshotId,
728        written: Option<Rc<dyn StateObject>>,
729    ) -> GlobalWrite {
730        assert!(
731            !global.has_pending_children(),
732            "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
733            global.pending_children(),
734            self.id,
735            snapshot_id
736        );
737        if let Some(state) = &written {
738            snapshot.record_write(Rc::clone(state));
739        }
740        mark_update_write(self.id);
741        let new_id = allocate_record_id();
742        let record = new_overwritable_record_as_head_locked(self.state);
743        GlobalWrite {
744            record,
745            new_id,
746            written,
747        }
748    }
749
750    fn finish_global_write(
751        &self,
752        global: &GlobalSnapshot,
753        write: GlobalWrite,
754        snapshot_id: SnapshotId,
755    ) {
756        // Published only now that its value is stored: nothing may read the
757        // record while the store runs.
758        write.record.set_snapshot_id(write.new_id);
759        write.record.set_tombstone(false);
760        advance_global_snapshot(write.new_id);
761        self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
762
763        let mut cleared: SmallVec<[Box<dyn RecordValue>; 2]> = SmallVec::new();
764        if !global.has_pending_children() {
765            let mut cursor = write.record.next();
766            while let Some(node) = cursor {
767                if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
768                    cleared.extend(node.clear_value());
769                    node.set_tombstone(true);
770                }
771                cursor = node.next();
772            }
773            self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
774        }
775
776        if let Some(modified) = write.written.as_ref() {
777            crate::snapshot_v2::notify_apply_observers(
778                std::slice::from_ref(modified),
779                write.new_id,
780            );
781        }
782        drop(cleared);
783    }
784
785    fn begin_child_write(
786        &self,
787        snapshot: &AnySnapshot,
788        snapshot_id: SnapshotId,
789        invalid: &SnapshotIdSet,
790        written: Option<Rc<dyn StateObject>>,
791    ) -> Rc<StateRecord> {
792        if let Some(state) = written {
793            snapshot.record_write(state);
794        }
795        mark_update_write(self.id);
796        self.writable_record(snapshot_id, invalid)
797    }
798
799    fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
800        let Some(readable) = readable_record_for(&self.head.clone_head(), snapshot_id, invalid)
801        else {
802            let current_head = self.head.clone_head();
803            let refreshed = readable_record_for(&current_head, snapshot_id, invalid);
804            let source = refreshed.unwrap_or_else(|| current_head.clone());
805
806            let cloned_value = source
807                .cloned_value()
808                .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"));
809            let new_head = StateRecord::with_boxed(snapshot_id, cloned_value, Some(current_head));
810            self.head.replace(new_head.clone());
811            self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
812            return new_head;
813        };
814
815        if readable.snapshot_id() == snapshot_id {
816            return readable;
817        }
818
819        let refreshed = {
820            let current_head = self.head.clone_head();
821            let refreshed = readable_record_for(&current_head, snapshot_id, invalid).unwrap_or_else(
822                || {
823                    panic!(
824                        "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
825                        self.id, snapshot_id, invalid
826                    )
827                },
828            );
829
830            if refreshed.snapshot_id() == snapshot_id {
831                return refreshed;
832            }
833
834            Rc::clone(&refreshed)
835        };
836
837        let overwritable = new_overwritable_record_locked(self.state);
838        if let Err(error) = overwritable.assign_value(&refreshed) {
839            log::error!(
840                "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
841                self.id,
842                error
843            );
844        }
845        overwritable.set_snapshot_id(snapshot_id);
846        overwritable.set_tombstone(false);
847
848        self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
849
850        overwritable
851    }
852}
853
854#[cold]
855#[inline(never)]
856fn check_chain_integrity(
857    head: &Rc<StateRecord>,
858    id: ObjectId,
859    caller: &str,
860    snapshot_context: Option<SnapshotId>,
861) {
862    let mut cursor = Some(Rc::clone(head));
863    let mut seen: HashSet<usize> = HashSet::default();
864    let mut ids = Vec::new();
865
866    while let Some(record) = cursor {
867        let addr = Rc::as_ptr(&record) as usize;
868        assert!(
869            seen.insert(addr),
870            "SnapshotMutableState::{caller} detected duplicate/cycle at record {:p} for state {id:?} (snapshot_context={snapshot_context:?}, chain_ids={ids:?})",
871            Rc::as_ptr(&record),
872        );
873        ids.push(record.snapshot_id());
874        cursor = record.next();
875    }
876
877    assert!(
878        !ids.is_empty(),
879        "SnapshotMutableState::{caller} finished integrity scan with empty id list for state {id:?} (snapshot_context={snapshot_context:?})"
880    );
881}
882
883fn should_check_chain_integrity() -> bool {
884    #[cfg(debug_assertions)]
885    {
886        true
887    }
888
889    #[cfg(not(debug_assertions))]
890    {
891        crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
892    }
893}
894
895impl<T: Clone + 'static> SnapshotMutableState<T> {
896    fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
897        let mut chain_ids = Vec::new();
898        let mut cursor = Some(self.first_record());
899        while let Some(record) = cursor {
900            chain_ids.push((record.snapshot_id(), record.is_tombstone()));
901            cursor = record.next();
902        }
903        chain_ids
904    }
905
906    fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
907        let snapshot = active_snapshot();
908        snapshot.record_read(self);
909
910        let snapshot_id = snapshot.snapshot_id();
911        if let Some(record) =
912            snapshot.with_invalid(|invalid| self.readable_for(snapshot_id, invalid))
913        {
914            return Ok(record);
915        }
916
917        let fresh_snapshot = active_snapshot();
918        let fresh_id = fresh_snapshot.snapshot_id();
919        let fresh_invalid = fresh_snapshot.invalid();
920
921        if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
922            return Ok(record);
923        }
924
925        let global = GlobalSnapshot::get_or_create();
926        let global_id = global.snapshot_id();
927        let global_invalid = global.invalid();
928
929        if let Some(record) = self.readable_for(global_id, &global_invalid) {
930            return Ok(record);
931        }
932
933        Err(StateReadFailure {
934            state_id: self.id,
935            snapshot_id,
936            fresh_snapshot_id: fresh_id,
937            fresh_invalid,
938            record_chain: self.record_chain_debug(),
939        })
940    }
941
942    fn readable_for(
943        &self,
944        snapshot_id: SnapshotId,
945        invalid: &SnapshotIdSet,
946    ) -> Option<Rc<StateRecord>> {
947        let head = self.first_record();
948        readable_record_for(&head, snapshot_id, invalid)
949    }
950
951    fn is_equivalent_to(&self, readable: Option<Rc<StateRecord>>, new_value: &T) -> bool {
952        readable.is_some_and(|record| {
953            record.with_value(|current: &T| self.policy.equivalent(current, new_value))
954        })
955    }
956
957    pub(crate) fn new_in_rc(initial: T, policy: Rc<dyn MutationPolicy<T>>) -> Rc<Self> {
958        let snapshot = active_snapshot();
959        let snapshot_id = snapshot.snapshot_id();
960
961        let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
962        let head = StateRecord::new(snapshot_id, initial, Some(tail));
963
964        Rc::new_cyclic(|weak_self| Self {
965            head: CurrentRecord::new(head),
966            policy,
967            id: ObjectId(weak_self.as_ptr() as usize),
968            weak_self: weak_self.clone(),
969            apply_observer: OnceCell::new(),
970            read_observation_lease: Rc::new(()),
971            scope_observation_count: Cell::new(0),
972            subscriber_callbacks: RefCell::new(Vec::new()),
973        })
974    }
975
976    pub(crate) fn set_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
977        assert!(
978            self.apply_observer.set(observer).is_ok(),
979            "state observer already installed"
980        );
981    }
982
983    fn acquire_observation_lease(&self) -> Option<Rc<dyn Any>> {
984        let was_empty = !self.has_subscribers();
985        let lease = Rc::clone(&self.read_observation_lease);
986        if was_empty {
987            notify_subscriber_callbacks(&self.subscriber_callbacks);
988        }
989        Some(lease)
990    }
991
992    fn add_scope_observer(&self) -> bool {
993        let was_empty = !self.has_subscribers();
994        let count = self
995            .scope_observation_count
996            .get()
997            .checked_add(1)
998            .expect("state scope observation count overflow");
999        self.scope_observation_count.set(count);
1000        was_empty
1001    }
1002
1003    fn remove_scope_observers(&self, count: usize) {
1004        if count == 0 {
1005            return;
1006        }
1007        let remaining = self
1008            .scope_observation_count
1009            .get()
1010            .checked_sub(count)
1011            .expect("state scope observation count underflow");
1012        self.scope_observation_count.set(remaining);
1013    }
1014
1015    fn has_subscribers(&self) -> bool {
1016        Rc::strong_count(&self.read_observation_lease) > 1 || self.scope_observation_count.get() > 0
1017    }
1018
1019    fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
1020        self.subscriber_callbacks
1021            .borrow_mut()
1022            .push(Rc::clone(&callback));
1023        if notify {
1024            callback();
1025        }
1026    }
1027
1028    fn notify_subscribers(&self) {
1029        notify_subscriber_callbacks(&self.subscriber_callbacks);
1030    }
1031
1032    fn finish_apply(&self, caller: &str, snapshot_id: SnapshotId) {
1033        finish_apply(
1034            &self.apply_observer,
1035            &self.head,
1036            self.id,
1037            caller,
1038            snapshot_id,
1039        );
1040    }
1041
1042    fn upgrade_self(&self) -> Option<Rc<Self>> {
1043        self.weak_self.upgrade()
1044    }
1045
1046    fn written_state(&self) -> Option<Rc<dyn StateObject>> {
1047        self.upgrade_self()
1048            .map(|state| -> Rc<dyn StateObject> { state })
1049    }
1050
1051    #[inline]
1052    pub(crate) fn id(&self) -> ObjectId {
1053        self.id
1054    }
1055
1056    pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1057        let record = self.readable_record_for_active_snapshot().ok()?;
1058        record.try_with_value(f)
1059    }
1060
1061    pub(crate) fn try_get(&self) -> Option<T> {
1062        self.try_with_value(Clone::clone)
1063    }
1064
1065    pub(crate) fn with_value<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1066        let record = self
1067            .readable_record_for_active_snapshot()
1068            .unwrap_or_else(|failure| panic!("{failure}"));
1069        record.with_value(f)
1070    }
1071
1072    pub(crate) fn get(&self) -> T {
1073        self.with_value(Clone::clone)
1074    }
1075
1076    pub(crate) fn set(&self, new_value: T) -> bool {
1077        #[cfg(debug_assertions)]
1078        {
1079            let in_handler = crate::in_event_handler();
1080            let in_snapshot = crate::in_applied_snapshot();
1081            if in_handler && !in_snapshot {
1082                log::warn!(
1083                    target: "cranpose::state",
1084                    "State modified in event handler without run_in_mutable_snapshot; \
1085                     this can make updates invisible to other contexts. Wrap the handler \
1086                     in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
1087                    self.id
1088                );
1089            }
1090        }
1091
1092        let snapshot = active_snapshot();
1093        let snapshot_id = snapshot.snapshot_id();
1094        let target = WriteTarget {
1095            head: &self.head,
1096            id: self.id,
1097            state: self,
1098        };
1099
1100        match &snapshot {
1101            AnySnapshot::Global(global) => {
1102                let readable =
1103                    snapshot.with_invalid(|invalid| self.readable_for(snapshot_id, invalid));
1104                if self.is_equivalent_to(readable, &new_value) {
1105                    return false;
1106                }
1107                let write =
1108                    target.begin_global_write(&snapshot, global, snapshot_id, self.written_state());
1109                let displaced = write.record.replace_value(new_value);
1110                target.finish_global_write(global, write, snapshot_id);
1111                drop(displaced);
1112            }
1113            AnySnapshot::Mutable(_)
1114            | AnySnapshot::NestedMutable(_)
1115            | AnySnapshot::TransparentMutable(_) => {
1116                let invalid = snapshot.invalid();
1117                if self.is_equivalent_to(self.readable_for(snapshot_id, &invalid), &new_value) {
1118                    return false;
1119                }
1120                let record = target.begin_child_write(
1121                    &snapshot,
1122                    snapshot_id,
1123                    &invalid,
1124                    self.written_state(),
1125                );
1126                let displaced = record.replace_value(new_value);
1127                target.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
1128                drop(displaced);
1129            }
1130            AnySnapshot::Readonly(_)
1131            | AnySnapshot::NestedReadonly(_)
1132            | AnySnapshot::TransparentReadonly(_) => {
1133                panic!("Cannot write to a read-only snapshot");
1134            }
1135        }
1136
1137        true
1138    }
1139}
1140
1141thread_local! {
1142    static ACTIVE_UPDATES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
1143    static PENDING_WRITES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
1144}
1145
1146pub(crate) struct UpdateScope {
1147    id: ObjectId,
1148    finished: bool,
1149}
1150
1151impl UpdateScope {
1152    pub(crate) fn new(id: ObjectId) -> Self {
1153        ACTIVE_UPDATES.with(|active| {
1154            active.borrow_mut().insert(id);
1155        });
1156        PENDING_WRITES.with(|pending| {
1157            pending.borrow_mut().remove(&id);
1158        });
1159        Self {
1160            id,
1161            finished: false,
1162        }
1163    }
1164
1165    pub(crate) fn finish(mut self) -> bool {
1166        self.finished = true;
1167        ACTIVE_UPDATES.with(|active| {
1168            active.borrow_mut().remove(&self.id);
1169        });
1170        PENDING_WRITES.with(|pending| pending.borrow_mut().remove(&self.id))
1171    }
1172}
1173
1174impl Drop for UpdateScope {
1175    fn drop(&mut self) {
1176        if self.finished {
1177            return;
1178        }
1179        ACTIVE_UPDATES.with(|active| {
1180            active.borrow_mut().remove(&self.id);
1181        });
1182        PENDING_WRITES.with(|pending| {
1183            pending.borrow_mut().remove(&self.id);
1184        });
1185    }
1186}
1187
1188fn mark_update_write(id: ObjectId) {
1189    ACTIVE_UPDATES.with(|active| {
1190        if active.borrow().contains(&id) {
1191            PENDING_WRITES.with(|pending| {
1192                pending.borrow_mut().insert(id);
1193            });
1194        }
1195    });
1196}
1197
1198impl<T: Clone + 'static> SnapshotMutableState<T> {
1199    fn try_readable_record(
1200        &self,
1201        snapshot_id: SnapshotId,
1202        invalid: &SnapshotIdSet,
1203    ) -> Option<Rc<StateRecord>> {
1204        self.readable_for(snapshot_id, invalid)
1205    }
1206}
1207
1208impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
1209    fn object_id(&self) -> ObjectId {
1210        self.id
1211    }
1212
1213    fn first_record(&self) -> Rc<StateRecord> {
1214        self.head.clone_head()
1215    }
1216
1217    fn try_readable_record(
1218        &self,
1219        snapshot_id: SnapshotId,
1220        invalid: &SnapshotIdSet,
1221    ) -> Option<Rc<StateRecord>> {
1222        self.try_readable_record(snapshot_id, invalid)
1223    }
1224
1225    fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
1226        self.try_readable_record(snapshot_id, invalid)
1227            .unwrap_or_else(|| {
1228                panic!(
1229                    "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
1230                    self.id, snapshot_id
1231                )
1232            })
1233    }
1234
1235    fn prepend_state_record(&self, record: Rc<StateRecord>) {
1236        self.head.prepend(record);
1237    }
1238
1239    fn observation_lease(&self) -> Option<Rc<dyn Any>> {
1240        self.acquire_observation_lease()
1241    }
1242
1243    fn merge_records(
1244        &self,
1245        previous: Rc<StateRecord>,
1246        current: Rc<StateRecord>,
1247        applied: Rc<StateRecord>,
1248    ) -> Option<Rc<StateRecord>> {
1249        let outcome = current
1250            .try_with_value(|current_value: &T| {
1251                applied
1252                    .try_with_value(|applied_value: &T| {
1253                        if self.policy.equivalent(current_value, applied_value) {
1254                            return Merge::KeepCurrent;
1255                        }
1256                        previous
1257                            .try_with_value(|previous_value: &T| {
1258                                Merge::Value(self.policy.merge(
1259                                    previous_value,
1260                                    current_value,
1261                                    applied_value,
1262                                ))
1263                            })
1264                            .unwrap_or(Merge::Missing("previous", &previous))
1265                    })
1266                    .unwrap_or(Merge::Missing("applied", &applied))
1267            })
1268            .unwrap_or(Merge::Missing("current", &current));
1269        match outcome {
1270            Merge::KeepCurrent => Some(current),
1271            Merge::Value(merged) => {
1272                merged.map(|value| StateRecord::new(applied.snapshot_id(), value, None))
1273            }
1274            Merge::Missing(which, record) => {
1275                log_merge_value_missing(which, self.id, record);
1276                None
1277            }
1278        }
1279    }
1280
1281    fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1282        promote_record_locked(self, child_id, TypeId::of::<T>())?;
1283        self.finish_apply("promote_record", child_id);
1284        Ok(())
1285    }
1286
1287    fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1288        let new_id = commit_merged_record_locked(self, &merged, TypeId::of::<T>())?;
1289        self.finish_apply("commit_merged_record", new_id);
1290        Ok(new_id)
1291    }
1292
1293    fn overwrite_unused_records(&self) -> bool {
1294        overwrite_unused_records_locked(self)
1295    }
1296
1297    fn as_any(&self) -> &dyn Any {
1298        self
1299    }
1300}
1301
1302pub(crate) struct MutableStateInner<T: Clone + 'static> {
1303    pub(crate) state: Rc<SnapshotMutableState<T>>,
1304    pub(crate) watchers: RefCell<HashMap<ScopeId, Weak<RecomposeScopeInner>>>,
1305    runtime: RuntimeHandle,
1306    state_id: Cell<Option<StateId>>,
1307}
1308
1309fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<Rc<dyn Fn()>>>) {
1310    let registered = callbacks.borrow().len();
1311    for index in 0..registered {
1312        let callback = callbacks.borrow().get(index).map(Rc::clone);
1313        if let Some(callback) = callback {
1314            callback();
1315        }
1316    }
1317}
1318
1319fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, Weak<RecomposeScopeInner>>) {
1320    let len = watchers.len();
1321    let capacity = watchers.capacity();
1322    if capacity > len.saturating_mul(4).max(32) {
1323        watchers.shrink_to_fit();
1324    }
1325}
1326
1327impl<T: Clone + 'static> MutableStateInner<T> {
1328    pub(crate) fn new_with_policy(
1329        value: T,
1330        runtime: RuntimeHandle,
1331        policy: Rc<dyn MutationPolicy<T>>,
1332    ) -> Self {
1333        Self {
1334            state: SnapshotMutableState::new_in_rc(value, policy),
1335            watchers: RefCell::new(HashMap::default()),
1336            runtime,
1337            state_id: Cell::new(None),
1338        }
1339    }
1340
1341    pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1342        self.state_id.set(Some(state_id));
1343        let runtime_handle = self.runtime.clone();
1344        self.state.set_apply_observer(Box::new(move || {
1345            let runtime = runtime_handle.clone();
1346            runtime_handle.enqueue_ui_task(Box::new(move || {
1347                runtime.with_state_arena(|arena| {
1348                    let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1349                        inner.invalidate_watchers();
1350                    });
1351                });
1352            }));
1353        }));
1354    }
1355
1356    /// Adds `scope` to the state's watchers, saying whether it was new and
1357    /// whether the state thereby gained its first subscriber.
1358    ///
1359    /// Dead watchers still count as subscribers until pruned. They are
1360    /// pruned whenever no live one is left, so a reader arriving after every
1361    /// earlier one died still makes the state newly subscribed, and
1362    /// otherwise only when the map is full, before the insert would grow
1363    /// it: a state read by thousands of scopes, such as the density, must
1364    /// not scan them all for every new reader.
1365    fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1366        let mut watchers = self.watchers.borrow_mut();
1367        let id = scope.id();
1368        match watchers.get(&id) {
1369            Some(existing) if existing.strong_count() > 0 => return (false, false),
1370            Some(_) => {
1371                watchers.remove(&id);
1372                self.state.remove_scope_observers(1);
1373            }
1374            None => {}
1375        }
1376        let any_live = watchers.values().any(|watcher| watcher.strong_count() > 0);
1377        if !any_live || watchers.len() >= watchers.capacity() {
1378            let before = watchers.len();
1379            watchers.retain(|_, watcher| watcher.strong_count() > 0);
1380            self.state.remove_scope_observers(before - watchers.len());
1381        }
1382        watchers.insert(id, scope.downgrade());
1383        drop(watchers);
1384        (true, self.state.add_scope_observer())
1385    }
1386
1387    fn has_subscribers(&self) -> bool {
1388        let mut watchers = self.watchers.borrow_mut();
1389        let before = watchers.len();
1390        watchers.retain(|_, existing| existing.upgrade().is_some());
1391        self.state.remove_scope_observers(before - watchers.len());
1392        self.state.has_subscribers()
1393    }
1394
1395    pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1396        let mut watchers = self.watchers.borrow_mut();
1397        let removed = if watchers
1398            .get(&scope_id)
1399            .is_some_and(|weak| weak.upgrade().is_none())
1400        {
1401            watchers.remove(&scope_id);
1402            shrink_watchers_if_sparse(&mut watchers);
1403            true
1404        } else {
1405            false
1406        };
1407        drop(watchers);
1408        self.state.remove_scope_observers(usize::from(removed));
1409    }
1410
1411    fn state_id(&self) -> Option<StateId> {
1412        self.state_id.get()
1413    }
1414
1415    fn invalidate_watchers(&self) {
1416        let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1417            let mut watchers = self.watchers.borrow_mut();
1418            let before = watchers.len();
1419            let mut live = Vec::with_capacity(watchers.len());
1420            watchers.retain(|_, scope| {
1421                if let Some(inner) = scope.upgrade() {
1422                    live.push(RecomposeScope { inner });
1423                    true
1424                } else {
1425                    false
1426                }
1427            });
1428            let removed_count = before - watchers.len();
1429            shrink_watchers_if_sparse(&mut watchers);
1430            (live, removed_count)
1431        };
1432        self.state.remove_scope_observers(removed_count);
1433
1434        for watcher in watchers {
1435            debug_record_scope_invalidation(
1436                watcher.id(),
1437                self.state_id.get(),
1438                std::any::type_name::<T>(),
1439            );
1440            if let Some(state_id) = self.state_id.get() {
1441                watcher.invalidate_from_state(state_id);
1442            } else {
1443                watcher.invalidate();
1444            }
1445        }
1446    }
1447}
1448
1449impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1450    fn drop(&mut self) {
1451        self.state
1452            .remove_scope_observers(self.watchers.get_mut().len());
1453    }
1454}
1455
1456fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1457    let Some(Some(scope)) =
1458        with_current_composer_opt(super::composer::Composer::current_state_invalidation_scope)
1459    else {
1460        return;
1461    };
1462    let (registered, became_subscribed) = inner.register_scope(&scope);
1463    if registered {
1464        if let Some(state_id) = inner.state_id() {
1465            scope.record_state_subscription(state_id);
1466        }
1467        if became_subscribed {
1468            inner.state.notify_subscribers();
1469        }
1470    }
1471}
1472
1473trait StateArenaHandle<T: Clone + 'static> {
1474    fn state_id(&self) -> StateId;
1475    fn runtime_id(&self) -> runtime::RuntimeId;
1476
1477    fn runtime_handle(&self) -> RuntimeHandle {
1478        runtime::runtime_handle_by_id(self.runtime_id())
1479            .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1480    }
1481
1482    fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1483        runtime::runtime_handle_by_id(self.runtime_id())
1484    }
1485
1486    fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1487        runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1488            arena.with_typed::<T, R>(self.state_id(), f)
1489        })
1490        .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1491    }
1492
1493    fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1494        runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1495            arena.with_typed_opt::<T, R>(self.state_id(), f)
1496        })?
1497    }
1498
1499    fn read_subscribed<R>(&self, read: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1500        self.with_inner(|inner| {
1501            let result = read(inner);
1502            register_current_state_scope(inner);
1503            result
1504        })
1505    }
1506}
1507
1508/// Cheap copyable read-only view of a state cell.
1509pub struct State<T: Clone + 'static> {
1510    id: StateId,
1511    runtime_id: runtime::RuntimeId,
1512    _marker: PhantomData<fn() -> T>,
1513}
1514
1515/// Cheap copyable mutable view of a state cell.
1516///
1517/// Ownership lives elsewhere: a composition slot, an [`OwnedMutableState`], or
1518/// the runtime for states created with [`crate::mutableStateOf`] /
1519/// [`MutableState::with_runtime`].
1520pub struct MutableState<T: Clone + 'static> {
1521    id: StateId,
1522    runtime_id: runtime::RuntimeId,
1523    _marker: PhantomData<fn() -> T>,
1524}
1525
1526/// Owning state handle for reclaimable state cells.
1527#[derive(Clone)]
1528pub struct OwnedMutableState<T: Clone + 'static> {
1529    state: MutableState<T>,
1530    _lease: Rc<runtime::StateHandleLease>,
1531    _marker: PhantomData<fn() -> T>,
1532}
1533
1534impl<T: Clone + 'static> PartialEq for State<T> {
1535    fn eq(&self, other: &Self) -> bool {
1536        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1537    }
1538}
1539
1540impl<T: Clone + 'static> Eq for State<T> {}
1541
1542impl<T: Clone + 'static> PartialEq for MutableState<T> {
1543    fn eq(&self, other: &Self) -> bool {
1544        self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1545    }
1546}
1547
1548/// Two owning handles are equal when they hold the same state, as two
1549/// [`MutableState`] handles are.
1550impl<T: Clone + 'static> PartialEq for OwnedMutableState<T> {
1551    fn eq(&self, other: &Self) -> bool {
1552        self.state == other.state
1553    }
1554}
1555
1556impl<T: Clone + 'static> Eq for MutableState<T> {}
1557
1558impl<T: Clone + 'static> Copy for State<T> {}
1559
1560impl<T: Clone + 'static> Clone for State<T> {
1561    fn clone(&self) -> Self {
1562        *self
1563    }
1564}
1565
1566impl<T: Clone + 'static> Copy for MutableState<T> {}
1567
1568impl<T: Clone + 'static> Clone for MutableState<T> {
1569    fn clone(&self) -> Self {
1570        *self
1571    }
1572}
1573
1574impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1575    fn state_id(&self) -> StateId {
1576        self.id
1577    }
1578
1579    fn runtime_id(&self) -> runtime::RuntimeId {
1580        self.runtime_id
1581    }
1582}
1583
1584impl<T: Clone + 'static> State<T> {
1585    pub fn is_alive(&self) -> bool {
1586        self.try_with_inner(|_| ()).is_some()
1587    }
1588
1589    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1590        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1591    }
1592
1593    pub fn try_value(&self) -> Option<T> {
1594        self.try_with_inner(|inner| inner.state.try_get())?
1595    }
1596
1597    /// Reads a copy of the value through `f` and subscribes the current
1598    /// scope. `f` may write this state; see [`Self::read`] for a read that
1599    /// borrows instead of copying.
1600    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1601        let value = self.read_subscribed(|inner| inner.state.get());
1602        f(&value)
1603    }
1604
1605    /// Reads the value in place through `f` and subscribes the current
1606    /// scope. Nothing is cloned, so a read of a large value costs nothing
1607    /// beyond `f`; in return `f` borrows the stored value and must not write
1608    /// this state.
1609    pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1610        self.read_subscribed(|inner| inner.state.with_value(f))
1611    }
1612
1613    pub fn value(&self) -> T {
1614        self.read_subscribed(|inner| inner.state.get())
1615    }
1616
1617    pub fn get(&self) -> T {
1618        self.value()
1619    }
1620
1621    pub fn has_subscribers(&self) -> bool {
1622        self.with_inner(MutableStateInner::has_subscribers)
1623    }
1624
1625    /// Runs `callback` each time this state gains its first subscriber, and
1626    /// right away when it already has one.
1627    ///
1628    /// The state owns `callback` for as long as the state lives, so nothing
1629    /// has to be kept alive on the caller's side.
1630    pub fn on_subscriber(&self, callback: impl Fn() + 'static) {
1631        let callback: Rc<dyn Fn()> = Rc::new(callback);
1632        self.with_inner(|inner| {
1633            inner
1634                .state
1635                .subscriber_callback(callback, inner.has_subscribers());
1636        });
1637    }
1638
1639    /// Counts this state as subscribed for as long as the returned guard
1640    /// lives, without observing it from any scope.
1641    ///
1642    /// Snapshot observers subscribe reads made under composition or the
1643    /// draw phase, and producers gated on [`State::has_subscribers`] — an
1644    /// infinite transition, for one — stop themselves when the last such
1645    /// subscriber leaves. A consumer that reads the value from a polling
1646    /// context (a frame effect stepping it into a derived state, a
1647    /// background task) is invisible to that accounting and would starve
1648    /// the producer it depends on. Holding this guard keeps the producer
1649    /// alive; dropping it releases the count.
1650    pub fn subscription_hold(&self) -> StateSubscriptionHold {
1651        StateSubscriptionHold {
1652            _lease: self
1653                .try_with_inner(|inner| inner.state.observation_lease())
1654                .flatten(),
1655        }
1656    }
1657}
1658
1659/// Keeps a [`State`] counted as subscribed while alive; see
1660/// [`State::subscription_hold`].
1661pub struct StateSubscriptionHold {
1662    _lease: Option<Rc<dyn Any>>,
1663}
1664
1665impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1666    fn state_id(&self) -> StateId {
1667        self.id
1668    }
1669
1670    fn runtime_id(&self) -> runtime::RuntimeId {
1671        self.runtime_id
1672    }
1673}
1674
1675impl<T: Clone + 'static> MutableState<T> {
1676    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1677        runtime.alloc_persistent_state(value)
1678    }
1679
1680    fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1681        Self {
1682            id,
1683            runtime_id,
1684            _marker: PhantomData,
1685        }
1686    }
1687
1688    pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1689        Self::from_parts(lease.id(), lease.runtime().id())
1690    }
1691
1692    pub fn is_alive(&self) -> bool {
1693        self.try_with_inner(|_| ()).is_some()
1694    }
1695
1696    pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1697        self.try_with_inner(|inner| inner.state.try_with_value(f))?
1698    }
1699
1700    pub fn try_value(&self) -> Option<T> {
1701        self.try_with_inner(|inner| inner.state.try_get())?
1702    }
1703
1704    pub fn as_state(&self) -> State<T> {
1705        State {
1706            id: self.id,
1707            runtime_id: self.runtime_id,
1708            _marker: PhantomData,
1709        }
1710    }
1711
1712    pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1713        let lease = self
1714            .runtime_handle_opt()?
1715            .retain_state_lease(self.state_id())?;
1716        Some(OwnedMutableState {
1717            state: *self,
1718            _lease: lease,
1719            _marker: PhantomData,
1720        })
1721    }
1722
1723    pub fn retain(&self) -> OwnedMutableState<T> {
1724        self.try_retain()
1725            .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1726    }
1727
1728    /// Reads a copy of the value through `f` and subscribes the current
1729    /// scope. `f` may write this state; see [`Self::read`] for a read that
1730    /// borrows instead of copying.
1731    pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1732        let value = self.read_subscribed(|inner| inner.state.get());
1733        f(&value)
1734    }
1735
1736    /// Reads the value in place through `f` and subscribes the current
1737    /// scope. Nothing is cloned, so a read of a large value costs nothing
1738    /// beyond `f`; in return `f` borrows the stored value and must not write
1739    /// this state.
1740    pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1741        self.read_subscribed(|inner| inner.state.with_value(f))
1742    }
1743
1744    pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1745        let runtime = self.runtime_handle();
1746        runtime.assert_ui_thread();
1747        runtime.with_state_arena(|arena| {
1748            arena.with_typed::<T, R>(self.state_id(), |inner| {
1749                let mut value = inner.state.get();
1750                let tracker = UpdateScope::new(inner.state.id());
1751                let result = f(&mut value);
1752                let wrote_elsewhere = tracker.finish();
1753                if !wrote_elsewhere && inner.state.set(value) {
1754                    inner.invalidate_watchers();
1755                }
1756                result
1757            })
1758        })
1759    }
1760
1761    pub fn replace(&self, value: T) {
1762        let Some(runtime) = self.runtime_handle_opt() else {
1763            log::debug!(
1764                "MutableState::replace skipped: runtime {:?} dropped",
1765                self.runtime_id()
1766            );
1767            return;
1768        };
1769        runtime.assert_ui_thread();
1770        let replaced = runtime
1771            .try_with_state_arena(|arena| {
1772                arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1773                    if inner.state.set(value) {
1774                        inner.invalidate_watchers();
1775                    }
1776                })
1777            })
1778            .flatten();
1779        if replaced.is_none() {
1780            log::debug!(
1781                "MutableState::replace skipped: state cell released (slot={}, gen={})",
1782                self.state_id().slot(),
1783                self.state_id().generation(),
1784            );
1785        }
1786    }
1787
1788    pub fn set_value(&self, value: T) {
1789        self.replace(value);
1790    }
1791
1792    pub fn set(&self, value: T) {
1793        self.replace(value);
1794    }
1795
1796    pub fn value(&self) -> T {
1797        self.read_subscribed(|inner| inner.state.get())
1798    }
1799
1800    pub fn get(&self) -> T {
1801        self.value()
1802    }
1803
1804    pub fn get_non_reactive(&self) -> T {
1805        self.with_inner(|inner| inner.state.get())
1806    }
1807
1808    #[doc(hidden)]
1809    pub fn runtime_state_id(&self) -> StateId {
1810        self.state_id()
1811    }
1812
1813    #[doc(hidden)]
1814    pub fn subscribe_current_scope_only(&self) {
1815        self.subscribe_current_scope();
1816    }
1817
1818    fn subscribe_current_scope(&self) {
1819        self.with_inner(register_current_state_scope::<T>);
1820    }
1821
1822    #[cfg(test)]
1823    pub(crate) fn watcher_count(&self) -> usize {
1824        self.with_inner(|inner| inner.watchers.borrow().len())
1825    }
1826
1827    #[cfg(test)]
1828    pub(crate) fn watcher_capacity(&self) -> usize {
1829        self.with_inner(|inner| inner.watchers.borrow().capacity())
1830    }
1831
1832    #[cfg(test)]
1833    pub(crate) fn state_id_for_test(&self) -> StateId {
1834        self.state_id()
1835    }
1836
1837    #[cfg(test)]
1838    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1839        self.as_state().subscribe_scope_for_test(scope);
1840    }
1841}
1842
1843impl<T: Clone + 'static> OwnedMutableState<T> {
1844    pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1845        let lease = runtime.alloc_state(value);
1846        Self {
1847            state: MutableState::from_lease(&lease),
1848            _lease: lease,
1849            _marker: PhantomData,
1850        }
1851    }
1852
1853    pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1854    where
1855        T: PartialEq,
1856    {
1857        Self::with_runtime_and_policy(value, runtime, Rc::new(StructuralEqual))
1858    }
1859
1860    pub(crate) fn with_runtime_and_policy(
1861        value: T,
1862        runtime: RuntimeHandle,
1863        policy: Rc<dyn MutationPolicy<T>>,
1864    ) -> Self {
1865        let lease = runtime.alloc_state_with_policy(value, policy);
1866        Self {
1867            state: MutableState::from_lease(&lease),
1868            _lease: lease,
1869            _marker: PhantomData,
1870        }
1871    }
1872
1873    pub fn handle(&self) -> MutableState<T> {
1874        self.state
1875    }
1876
1877    pub fn as_state(&self) -> State<T> {
1878        self.state.as_state()
1879    }
1880}
1881
1882impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1883    type Target = MutableState<T>;
1884
1885    fn deref(&self) -> &Self::Target {
1886        &self.state
1887    }
1888}
1889
1890#[cfg(test)]
1891impl<T: Clone + 'static> State<T> {
1892    pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1893        self.with_inner(|inner| {
1894            let (registered, became_subscribed) = inner.register_scope(scope);
1895            if registered {
1896                if let Some(state_id) = inner.state_id() {
1897                    scope.record_state_subscription(state_id);
1898                }
1899                if became_subscribed {
1900                    inner.state.notify_subscribers();
1901                }
1902            }
1903        });
1904    }
1905}
1906
1907impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1908    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1909        if let Some(value) = self.try_value() {
1910            f.debug_struct("MutableState")
1911                .field("value", &value)
1912                .finish()
1913        } else {
1914            f.write_str("MutableState { value: <unavailable> }")
1915        }
1916    }
1917}
1918
1919#[derive(Clone)]
1920pub struct SnapshotStateList<T: Clone + 'static> {
1921    state: OwnedMutableState<Vec<T>>,
1922}
1923
1924impl<T: Clone + 'static> SnapshotStateList<T> {
1925    pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1926    where
1927        I: IntoIterator<Item = T>,
1928    {
1929        let initial: Vec<T> = values.into_iter().collect();
1930        Self {
1931            state: OwnedMutableState::with_runtime(initial, runtime),
1932        }
1933    }
1934
1935    pub fn as_state(&self) -> State<Vec<T>> {
1936        self.state.as_state()
1937    }
1938
1939    pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1940        self.state.handle()
1941    }
1942
1943    pub fn len(&self) -> usize {
1944        self.state.with(Vec::len)
1945    }
1946
1947    pub fn is_empty(&self) -> bool {
1948        self.len() == 0
1949    }
1950
1951    pub fn to_vec(&self) -> Vec<T> {
1952        self.state.with(Clone::clone)
1953    }
1954
1955    pub fn iter(&self) -> Vec<T> {
1956        self.to_vec()
1957    }
1958
1959    pub fn get(&self, index: usize) -> T {
1960        self.state.with(|values| values[index].clone())
1961    }
1962
1963    pub fn get_opt(&self, index: usize) -> Option<T> {
1964        self.state.with(|values| values.get(index).cloned())
1965    }
1966
1967    pub fn first(&self) -> Option<T> {
1968        self.get_opt(0)
1969    }
1970
1971    pub fn last(&self) -> Option<T> {
1972        self.state.with(|values| values.last().cloned())
1973    }
1974
1975    pub fn push(&self, value: T) {
1976        self.state.update(|values| values.push(value));
1977    }
1978
1979    pub fn extend<I>(&self, iter: I)
1980    where
1981        I: IntoIterator<Item = T>,
1982    {
1983        self.state.update(|values| values.extend(iter));
1984    }
1985
1986    pub fn insert(&self, index: usize, value: T) {
1987        self.state.update(|values| values.insert(index, value));
1988    }
1989
1990    pub fn set(&self, index: usize, value: T) -> T {
1991        self.state
1992            .update(|values| std::mem::replace(&mut values[index], value))
1993    }
1994
1995    pub fn remove(&self, index: usize) -> T {
1996        self.state.update(|values| values.remove(index))
1997    }
1998
1999    pub fn pop(&self) -> Option<T> {
2000        self.state.update(Vec::pop)
2001    }
2002
2003    pub fn clear(&self) {
2004        self.state.replace(Vec::new());
2005    }
2006
2007    pub fn retain<F>(&self, mut predicate: F)
2008    where
2009        F: FnMut(&T) -> bool,
2010    {
2011        self.state
2012            .update(|values| values.retain(|value| predicate(value)));
2013    }
2014
2015    pub fn replace_with<I>(&self, iter: I)
2016    where
2017        I: IntoIterator<Item = T>,
2018    {
2019        self.state.replace(iter.into_iter().collect());
2020    }
2021}
2022
2023impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
2024    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2025        let contents = self.to_vec();
2026        f.debug_struct("SnapshotStateList")
2027            .field("values", &contents)
2028            .finish()
2029    }
2030}
2031
2032#[derive(Clone)]
2033pub struct SnapshotStateMap<K, V>
2034where
2035    K: Clone + Eq + Hash + 'static,
2036    V: Clone + 'static,
2037{
2038    state: OwnedMutableState<HashMap<K, V>>,
2039}
2040
2041impl<K, V> SnapshotStateMap<K, V>
2042where
2043    K: Clone + Eq + Hash + 'static,
2044    V: Clone + 'static,
2045{
2046    pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
2047    where
2048        I: IntoIterator<Item = (K, V)>,
2049    {
2050        let map: HashMap<K, V> = pairs.into_iter().collect();
2051        Self {
2052            state: OwnedMutableState::with_runtime(map, runtime),
2053        }
2054    }
2055
2056    pub fn as_state(&self) -> State<HashMap<K, V>> {
2057        self.state.as_state()
2058    }
2059
2060    pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
2061        self.state.handle()
2062    }
2063
2064    pub fn len(&self) -> usize {
2065        self.state.with(std::collections::HashMap::len)
2066    }
2067
2068    pub fn is_empty(&self) -> bool {
2069        self.state.with(std::collections::HashMap::is_empty)
2070    }
2071
2072    pub fn contains_key(&self, key: &K) -> bool {
2073        self.state.with(|map| map.contains_key(key))
2074    }
2075
2076    pub fn get(&self, key: &K) -> Option<V> {
2077        self.state.with(|map| map.get(key).cloned())
2078    }
2079
2080    pub fn to_hash_map(&self) -> HashMap<K, V> {
2081        self.state.with(Clone::clone)
2082    }
2083
2084    pub fn insert(&self, key: K, value: V) -> Option<V> {
2085        self.state.update(|map| map.insert(key, value))
2086    }
2087
2088    pub fn extend<I>(&self, iter: I)
2089    where
2090        I: IntoIterator<Item = (K, V)>,
2091    {
2092        self.state.update(|map| map.extend(iter));
2093    }
2094
2095    pub fn remove(&self, key: &K) -> Option<V> {
2096        self.state.update(|map| map.remove(key))
2097    }
2098
2099    pub fn clear(&self) {
2100        self.state.replace(HashMap::default());
2101    }
2102
2103    pub fn retain<F>(&self, mut predicate: F)
2104    where
2105        F: FnMut(&K, &mut V) -> bool,
2106    {
2107        self.state.update(|map| map.retain(|k, v| predicate(k, v)));
2108    }
2109}
2110
2111impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
2112where
2113    K: Clone + Eq + Hash + fmt::Debug + 'static,
2114    V: Clone + fmt::Debug + 'static,
2115{
2116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2117        let contents = self.to_hash_map();
2118        f.debug_struct("SnapshotStateMap")
2119            .field("entries", &contents)
2120            .finish()
2121    }
2122}
2123
2124pub(crate) struct DerivedState<T: Clone + PartialEq + 'static> {
2125    compute: Rc<dyn Fn() -> T>,
2126    pub(crate) state: OwnedMutableState<T>,
2127}
2128
2129impl<T: Clone + PartialEq + 'static> DerivedState<T> {
2130    pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
2131        let initial = compute();
2132        Self {
2133            compute,
2134            state: OwnedMutableState::with_runtime_structural_eq(initial, runtime),
2135        }
2136    }
2137
2138    pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
2139        self.compute = compute;
2140    }
2141
2142    pub(crate) fn recompute(&self) {
2143        let value = (self.compute)();
2144        self.state.set_value(value);
2145    }
2146}
2147
2148impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
2149    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2150        if let Some(value) = self.try_value() {
2151            f.debug_struct("State").field("value", &value).finish()
2152        } else {
2153            f.write_str("State { value: <unavailable> }")
2154        }
2155    }
2156}
2157
2158#[cfg(test)]
2159#[path = "tests/state_tests.rs"]
2160mod tests;