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(¤t_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(¤t_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(¤t_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, ¤t_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
1327pub struct State<T: Clone + 'static> {
1329 id: StateId,
1330 runtime_id: runtime::RuntimeId,
1331 _marker: PhantomData<fn() -> T>,
1332}
1333
1334pub struct MutableState<T: Clone + 'static> {
1340 id: StateId,
1341 runtime_id: runtime::RuntimeId,
1342 _marker: PhantomData<fn() -> T>,
1343}
1344
1345#[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 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 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 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
1473pub 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 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 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;