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