Skip to main content

cranpose_core/
state.rs

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