1#![allow(clippy::arc_with_non_send_sync)]
3
4use crate::collections::map::{HashMap, HashSet};
5use crate::debug_trace::debug_record_scope_invalidation;
6use std::any::Any;
7use std::cell::{Cell, RefCell};
8use std::fmt;
9use std::hash::Hash;
10use std::marker::PhantomData;
11use std::ops::Deref;
12use std::rc::{Rc, Weak as RcWeak};
13use std::sync::{Arc, Mutex, MutexGuard, Weak};
14
15use crate::snapshot_id_set::{SnapshotId, SnapshotIdSet};
16use crate::snapshot_pinning::lowest_pinned_snapshot;
17use crate::snapshot_v2::{
18 advance_global_snapshot, allocate_record_id, current_snapshot, AnySnapshot, GlobalSnapshot,
19};
20use crate::{
21 runtime, with_current_composer_opt, RecomposeScope, RecomposeScopeInner, RuntimeHandle,
22 ScopeId, StateId,
23};
24
25pub(crate) const PREEXISTING_SNAPSHOT_ID: SnapshotId = 1;
26
27const INVALID_SNAPSHOT_ID: SnapshotId = 0;
28
29const SNAPSHOT_ID_MAX: SnapshotId = usize::MAX;
31
32#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug, Default)]
33pub struct ObjectId(pub(crate) usize);
34
35impl ObjectId {
36 pub(crate) fn new<T: ?Sized + 'static>(object: &Arc<T>) -> Self {
37 Self(Arc::as_ptr(object) as *const () as usize)
38 }
39
40 #[inline]
41 pub(crate) fn as_usize(self) -> usize {
42 self.0
43 }
44}
45
46pub struct StateRecord {
53 snapshot_id: Cell<SnapshotId>,
54 tombstone: Cell<bool>,
55 next: Cell<Option<Rc<StateRecord>>>,
56 value: RefCell<Option<Box<dyn Any>>>,
57}
58
59#[derive(Debug)]
60struct StateReadFailure {
61 state_id: ObjectId,
62 snapshot_id: SnapshotId,
63 fresh_snapshot_id: SnapshotId,
64 fresh_invalid: SnapshotIdSet,
65 record_chain: Vec<(SnapshotId, bool)>,
66}
67
68impl std::fmt::Display for StateReadFailure {
69 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
70 write!(
71 f,
72 "Reading a state that was created after the snapshot was taken or in a snapshot that has not yet been applied\n\
73 state={:?}, snapshot_id={}, fresh_snapshot_id={}, fresh_invalid={:?}\n\
74 record_chain={:?}",
75 self.state_id,
76 self.snapshot_id,
77 self.fresh_snapshot_id,
78 self.fresh_invalid,
79 self.record_chain
80 )
81 }
82}
83
84#[derive(Debug, Clone, Copy, Eq, PartialEq)]
85pub(crate) enum StateRecordValueError {
86 MissingOrWrongType { expected: &'static str },
87}
88
89impl StateRecord {
90 pub(crate) fn new<T: Any>(
91 snapshot_id: SnapshotId,
92 value: T,
93 next: Option<Rc<StateRecord>>,
94 ) -> Rc<Self> {
95 Rc::new(Self {
96 snapshot_id: Cell::new(snapshot_id),
97 tombstone: Cell::new(false),
98 next: Cell::new(next),
99 value: RefCell::new(Some(Box::new(value))),
100 })
101 }
102
103 #[inline]
104 pub(crate) fn snapshot_id(&self) -> SnapshotId {
105 self.snapshot_id.get()
106 }
107
108 #[inline]
109 pub(crate) fn set_snapshot_id(&self, id: SnapshotId) {
110 self.snapshot_id.set(id);
111 }
112
113 #[inline]
114 pub(crate) fn next(&self) -> Option<Rc<StateRecord>> {
115 self.next.take().inspect(|record| {
116 self.next.set(Some(Rc::clone(record)));
117 })
118 }
119
120 #[inline]
121 pub(crate) fn set_next(&self, next: Option<Rc<StateRecord>>) {
122 self.next.set(next);
123 }
124
125 #[inline]
126 pub(crate) fn is_tombstone(&self) -> bool {
127 self.tombstone.get()
128 }
129
130 #[inline]
131 pub(crate) fn set_tombstone(&self, tombstone: bool) {
132 self.tombstone.set(tombstone);
133 }
134
135 pub(crate) fn clear_value(&self) {
136 self.value.borrow_mut().take();
137 }
138
139 pub(crate) fn replace_value<T: Any>(&self, new_value: T) {
140 *self.value.borrow_mut() = Some(Box::new(new_value));
141 }
142
143 pub(crate) fn with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> R {
144 self.try_with_value(f)
145 .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"))
146 }
147
148 pub(crate) fn try_with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
149 let guard = self.value.borrow();
150 let value = guard.as_ref().and_then(|boxed| boxed.downcast_ref::<T>())?;
151 Some(f(value))
152 }
153
154 #[cfg(test)]
157 pub(crate) fn clear_for_reuse(&self) {
158 self.clear_value();
159 }
160
161 pub(crate) fn assign_value<T: Any + Clone>(
168 &self,
169 source: &StateRecord,
170 ) -> Result<(), StateRecordValueError> {
171 let cloned_value = source.try_with_value(|value: &T| value.clone()).ok_or(
172 StateRecordValueError::MissingOrWrongType {
173 expected: std::any::type_name::<T>(),
174 },
175 )?;
176 self.replace_value(cloned_value);
177 Ok(())
178 }
179}
180
181impl Drop for StateRecord {
182 fn drop(&mut self) {
183 let mut next = self.next.take();
186 while let Some(node) = next {
187 match Rc::try_unwrap(node) {
188 Ok(record) => {
189 next = record.next.take();
193 }
194 Err(_) => {
195 break;
198 }
199 }
200 }
201 }
202}
203
204struct CurrentRecord {
211 head: RefCell<Rc<StateRecord>>,
212}
213
214impl CurrentRecord {
215 fn new(head: Rc<StateRecord>) -> Self {
216 Self {
217 head: RefCell::new(head),
218 }
219 }
220
221 fn clone_head(&self) -> Rc<StateRecord> {
222 self.head.borrow().clone()
223 }
224
225 fn replace(&self, new_head: Rc<StateRecord>) {
226 *self.head.borrow_mut() = new_head;
227 }
228
229 fn prepend(&self, record: Rc<StateRecord>) {
230 let current_head = self.clone_head();
231 record.set_next(Some(current_head));
232 self.replace(record);
233 }
234}
235
236#[inline]
237fn record_is_valid_for(
238 record: &Rc<StateRecord>,
239 snapshot_id: SnapshotId,
240 invalid: &SnapshotIdSet,
241) -> bool {
242 if record.is_tombstone() {
243 return false;
244 }
245
246 let candidate = record.snapshot_id();
247 if candidate == INVALID_SNAPSHOT_ID || candidate > snapshot_id {
248 return false;
249 }
250
251 candidate == snapshot_id || !invalid.get(candidate)
252}
253
254pub(crate) fn readable_record_for(
255 head: &Rc<StateRecord>,
256 snapshot_id: SnapshotId,
257 invalid: &SnapshotIdSet,
258) -> Option<Rc<StateRecord>> {
259 let mut best: Option<Rc<StateRecord>> = None;
263 let mut cursor = Some(Rc::clone(head));
264
265 while let Some(record) = cursor {
266 if record_is_valid_for(&record, snapshot_id, invalid) {
267 let replace = best
268 .as_ref()
269 .map(|current| current.snapshot_id() < record.snapshot_id())
270 .unwrap_or(true);
271 if replace {
272 best = Some(Rc::clone(&record));
273 }
274 }
275 cursor = record.next();
276 }
277
278 best
279}
280
281fn find_youngest_or<F>(head: &Rc<StateRecord>, predicate: F) -> Rc<StateRecord>
287where
288 F: Fn(&Rc<StateRecord>) -> bool,
289{
290 let mut current = Some(Rc::clone(head));
291 let mut youngest = Rc::clone(head);
292
293 while let Some(record) = current {
294 if predicate(&record) {
295 return record;
296 }
297 if youngest.snapshot_id() < record.snapshot_id() {
298 youngest = Rc::clone(&record);
299 }
300 current = record.next();
301 }
302
303 youngest
304}
305
306pub(crate) fn used_locked(head: &Rc<StateRecord>) -> Option<Rc<StateRecord>> {
318 let mut current = Some(Rc::clone(head));
319 let mut valid_record: Option<Rc<StateRecord>> = None;
320
321 let reuse_limit = lowest_pinned_snapshot()
323 .map(|lowest| lowest.saturating_sub(1))
324 .unwrap_or_else(|| allocate_record_id().saturating_sub(1));
325
326 let invalid = SnapshotIdSet::EMPTY;
327
328 while let Some(record) = current {
329 let current_id = record.snapshot_id();
330
331 if current_id == PREEXISTING_SNAPSHOT_ID {
333 current = record.next();
334 continue;
335 }
336
337 if current_id == INVALID_SNAPSHOT_ID {
339 return Some(record);
340 }
341
342 if record.is_tombstone() && current_id < reuse_limit {
343 return Some(record);
344 }
345
346 if record_is_valid_for(&record, reuse_limit, &invalid) {
348 if let Some(ref existing) = valid_record {
349 return Some(if current_id < existing.snapshot_id() {
352 record
353 } else {
354 Rc::clone(existing)
355 });
356 } else {
357 valid_record = Some(record.clone());
359 }
360 }
361
362 current = record.next();
363 }
364
365 None
367}
368
369pub(crate) fn new_overwritable_record_locked(state: &dyn StateObject) -> Rc<StateRecord> {
380 let state_head = state.first_record();
381
382 if let Some(reusable) = used_locked(&state_head) {
384 reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
386 return reusable;
387 }
388
389 let new_record = StateRecord::new(
392 SNAPSHOT_ID_MAX,
393 (),
394 None, );
396
397 state.prepend_state_record(Rc::clone(&new_record));
399
400 new_record
401}
402
403pub(crate) fn new_overwritable_record_as_head_locked(state: &dyn StateObject) -> Rc<StateRecord> {
409 let head = state.first_record();
410
411 if let Some(reusable) = used_locked(&head) {
412 reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
413
414 if !Rc::ptr_eq(&head, &reusable) {
415 let mut cursor = Some(Rc::clone(&head));
416 let mut unlinked = false;
417
418 while let Some(node) = cursor {
419 let next = node.next();
420 if let Some(next_record) = next {
421 if Rc::ptr_eq(&next_record, &reusable) {
422 node.set_next(reusable.next());
423 unlinked = true;
424 break;
425 }
426 cursor = Some(next_record);
427 } else {
428 break;
429 }
430 }
431
432 if !unlinked {
433 debug_assert!(
434 false,
435 "new_overwritable_record_as_head_locked: reusable record not found in chain"
436 );
437 let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
438 state.prepend_state_record(Rc::clone(&new_record));
439 return new_record;
440 }
441
442 state.prepend_state_record(Rc::clone(&reusable));
443 }
444
445 return reusable;
446 }
447
448 let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
449 state.prepend_state_record(Rc::clone(&new_record));
450 new_record
451}
452
453pub(crate) fn overwrite_unused_records_locked<T: Any + Clone>(state: &dyn StateObject) -> bool {
467 let head = state.first_record();
468 let mut current = Some(Rc::clone(&head));
469 let mut overwrite_record: Option<Rc<StateRecord>> = None;
470 let mut valid_record: Option<Rc<StateRecord>> = None;
471
472 let reuse_limit =
475 lowest_pinned_snapshot().unwrap_or_else(crate::snapshot_v2::peek_next_snapshot_id);
476
477 let mut retained_records = 0;
478
479 while let Some(record) = current {
480 let current_id = record.snapshot_id();
481
482 if current_id == INVALID_SNAPSHOT_ID {
483 } else if current_id < reuse_limit {
485 if valid_record.is_none() {
486 valid_record = Some(Rc::clone(&record));
489 retained_records += 1;
490 } else {
491 let Some(valid) = valid_record.as_ref() else {
494 valid_record = Some(Rc::clone(&record));
495 retained_records += 1;
496 current = record.next();
497 continue;
498 };
499 let record_to_overwrite = if current_id < valid.snapshot_id() {
500 Rc::clone(&record)
501 } else {
502 let to_overwrite = Rc::clone(valid);
504 valid_record = Some(Rc::clone(&record));
505 to_overwrite
506 };
507
508 let source_record = overwrite_record.get_or_insert_with(|| {
510 find_youngest_or(&head, |r| r.snapshot_id() >= reuse_limit)
511 });
512
513 record_to_overwrite.set_snapshot_id(INVALID_SNAPSHOT_ID);
515 if let Err(error) = record_to_overwrite.assign_value::<T>(source_record) {
516 log::error!(
517 "snapshot cleanup could not copy retained state record value for state {:?}: {:?}",
518 state.object_id(),
519 error
520 );
521 }
522 }
523 } else {
524 retained_records += 1;
526 }
527
528 current = record.next();
529 }
530
531 retained_records > 1
534}
535
536fn active_snapshot() -> AnySnapshot {
537 current_snapshot().unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()))
538}
539
540pub(crate) trait MutationPolicy<T>: Send + Sync {
541 fn equivalent(&self, a: &T, b: &T) -> bool;
542 fn merge(&self, _previous: &T, _current: &T, _applied: &T) -> Option<T> {
543 None
544 }
545}
546
547pub(crate) struct NeverEqual;
548
549impl<T> MutationPolicy<T> for NeverEqual {
550 fn equivalent(&self, _a: &T, _b: &T) -> bool {
551 false
552 }
553}
554
555pub(crate) struct StructuralEqual;
556
557impl<T: PartialEq> MutationPolicy<T> for StructuralEqual {
558 fn equivalent(&self, a: &T, b: &T) -> bool {
559 a == b
560 }
561}
562
563pub trait StateObject: Any {
564 fn object_id(&self) -> ObjectId;
565 fn first_record(&self) -> Rc<StateRecord>;
566 fn try_readable_record(
567 &self,
568 snapshot_id: SnapshotId,
569 invalid: &SnapshotIdSet,
570 ) -> Option<Rc<StateRecord>>;
571 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord>;
572
573 fn prepend_state_record(&self, record: Rc<StateRecord>);
577
578 fn merge_records(
579 &self,
580 _previous: Rc<StateRecord>,
581 _current: Rc<StateRecord>,
582 _applied: Rc<StateRecord>,
583 ) -> Option<Rc<StateRecord>> {
584 None
585 }
586
587 fn commit_merged_record(&self, _merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
588 Err("StateObject does not support merged record commits")
589 }
590 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str>;
591
592 fn overwrite_unused_records(&self) -> bool {
597 false }
599
600 fn observation_lease(&self) -> Option<Box<dyn Any>> {
601 None
602 }
603
604 fn as_any(&self) -> &dyn Any;
606}
607
608pub(crate) struct SnapshotMutableState<T> {
609 head: CurrentRecord,
610 policy: Arc<dyn MutationPolicy<T>>,
611 id: ObjectId,
612 weak_self: Mutex<Option<Weak<Self>>>,
613 apply_observers: Mutex<Vec<Box<dyn Fn() + 'static>>>,
614 read_observation_count: Cell<usize>,
615 scope_observation_count: Cell<usize>,
616 subscriber_callbacks: RefCell<Vec<RcWeak<dyn Fn()>>>,
617}
618
619struct StateObservationLease<T: Clone + 'static> {
620 state: Weak<SnapshotMutableState<T>>,
621}
622
623impl<T: Clone + 'static> Drop for StateObservationLease<T> {
624 fn drop(&mut self) {
625 if let Some(state) = self.state.upgrade() {
626 let count = state
627 .read_observation_count
628 .get()
629 .checked_sub(1)
630 .expect("state observation lease count underflow");
631 state.read_observation_count.set(count);
632 }
633 }
634}
635
636impl<T> SnapshotMutableState<T> {
637 fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
638 if !should_check_chain_integrity() {
639 return;
640 }
641 let head = self.head.clone_head();
642 let mut cursor = Some(head);
643 let mut seen: HashSet<usize> = HashSet::default();
644 let mut ids = Vec::new();
645
646 while let Some(record) = cursor {
647 let addr = Rc::as_ptr(&record) as usize;
648 assert!(
649 seen.insert(addr),
650 "SnapshotMutableState::{} detected duplicate/cycle at record {:p} for state {:?} (snapshot_context={:?}, chain_ids={:?})",
651 caller,
652 Rc::as_ptr(&record),
653 self.id,
654 snapshot_context,
655 ids
656 );
657 ids.push(record.snapshot_id());
658 cursor = record.next();
659 }
660
661 assert!(
662 !ids.is_empty(),
663 "SnapshotMutableState::{} finished integrity scan with empty id list for state {:?} (snapshot_context={:?})",
664 caller,
665 self.id,
666 snapshot_context
667 );
668 }
669}
670
671fn should_check_chain_integrity() -> bool {
672 #[cfg(debug_assertions)]
673 {
674 true
675 }
676
677 #[cfg(not(debug_assertions))]
678 {
679 crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
680 }
681}
682
683impl<T: Clone + 'static> SnapshotMutableState<T> {
684 fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
685 let mut chain_ids = Vec::new();
686 let mut cursor = Some(self.first_record());
687 while let Some(record) = cursor {
688 chain_ids.push((record.snapshot_id(), record.is_tombstone()));
689 cursor = record.next();
690 }
691 chain_ids
692 }
693
694 fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
695 let snapshot = active_snapshot();
696 if let Some(state) = self.upgrade_self() {
697 snapshot.record_read(&*state);
698 }
699
700 let snapshot_id = snapshot.snapshot_id();
701 let invalid = snapshot.invalid();
702
703 if let Some(record) = self.readable_for(snapshot_id, &invalid) {
704 return Ok(record);
705 }
706
707 let fresh_snapshot = active_snapshot();
708 let fresh_id = fresh_snapshot.snapshot_id();
709 let fresh_invalid = fresh_snapshot.invalid();
710
711 if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
712 return Ok(record);
713 }
714
715 let global = GlobalSnapshot::get_or_create();
716 let global_id = global.snapshot_id();
717 let global_invalid = global.invalid();
718
719 if let Some(record) = self.readable_for(global_id, &global_invalid) {
720 return Ok(record);
721 }
722
723 Err(StateReadFailure {
724 state_id: self.id,
725 snapshot_id,
726 fresh_snapshot_id: fresh_id,
727 fresh_invalid,
728 record_chain: self.record_chain_debug(),
729 })
730 }
731
732 fn readable_for(
733 &self,
734 snapshot_id: SnapshotId,
735 invalid: &SnapshotIdSet,
736 ) -> Option<Rc<StateRecord>> {
737 let head = self.first_record();
738 readable_record_for(&head, snapshot_id, invalid)
739 }
740
741 fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
742 let readable = match self.readable_for(snapshot_id, invalid) {
743 Some(record) => record,
744 None => {
745 let current_head = self.head.clone_head();
746 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid);
747 let source = refreshed.unwrap_or_else(|| current_head.clone());
748
749 let cloned_value = source.with_value(|value: &T| value.clone());
753 let new_head = StateRecord::new(snapshot_id, cloned_value, Some(current_head));
754 self.head.replace(new_head.clone());
755 self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
756 return new_head;
757 }
758 };
759
760 if readable.snapshot_id() == snapshot_id {
761 return readable;
762 }
763
764 let refreshed = {
765 let current_head = self.head.clone_head();
766 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid).unwrap_or_else(
767 || {
768 panic!(
769 "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
770 self.id, snapshot_id, invalid
771 )
772 },
773 );
774
775 if refreshed.snapshot_id() == snapshot_id {
776 return refreshed;
777 }
778
779 Rc::clone(&refreshed)
780 };
781
782 let overwritable = new_overwritable_record_locked(self);
783 if let Err(error) = overwritable.assign_value::<T>(&refreshed) {
784 log::error!(
785 "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
786 self.id,
787 error
788 );
789 }
790 overwritable.set_snapshot_id(snapshot_id);
791 overwritable.set_tombstone(false);
792
793 self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
794
795 overwritable
796 }
797
798 pub(crate) fn new_in_arc(initial: T, policy: Arc<dyn MutationPolicy<T>>) -> Arc<Self> {
799 let snapshot = active_snapshot();
800 let snapshot_id = snapshot.snapshot_id();
801
802 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
803 let head = StateRecord::new(snapshot_id, initial, Some(tail));
804
805 let mut state = Arc::new(Self {
806 head: CurrentRecord::new(head),
807 policy,
808 id: ObjectId::default(),
809 weak_self: Mutex::new(None),
810 apply_observers: Mutex::new(Vec::new()),
811 read_observation_count: Cell::new(0),
812 scope_observation_count: Cell::new(0),
813 subscriber_callbacks: RefCell::new(Vec::new()),
814 });
815
816 let id = ObjectId::new(&state);
817 if let Some(state_inner) = Arc::get_mut(&mut state) {
818 state_inner.id = id;
819 }
820
821 *state.lock_weak_self() = Some(Arc::downgrade(&state));
822
823 state
826 }
827
828 pub(crate) fn add_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
829 self.lock_apply_observers().push(observer);
830 }
831
832 fn acquire_observation_lease(&self) -> Option<Box<dyn Any>> {
833 let state = self.lock_weak_self().as_ref()?.clone();
834 let was_empty = !self.has_subscribers();
835 let count = self
836 .read_observation_count
837 .get()
838 .checked_add(1)
839 .expect("state observation lease count overflow");
840 self.read_observation_count.set(count);
841 if was_empty {
842 notify_subscriber_callbacks(&self.subscriber_callbacks);
843 }
844 Some(Box::new(StateObservationLease { state }))
845 }
846
847 fn add_scope_observer(&self) -> bool {
848 let was_empty = !self.has_subscribers();
849 let count = self
850 .scope_observation_count
851 .get()
852 .checked_add(1)
853 .expect("state scope observation count overflow");
854 self.scope_observation_count.set(count);
855 was_empty
856 }
857
858 fn remove_scope_observers(&self, count: usize) {
859 if count == 0 {
860 return;
861 }
862 let remaining = self
863 .scope_observation_count
864 .get()
865 .checked_sub(count)
866 .expect("state scope observation count underflow");
867 self.scope_observation_count.set(remaining);
868 }
869
870 fn has_subscribers(&self) -> bool {
871 self.read_observation_count.get() > 0 || self.scope_observation_count.get() > 0
872 }
873
874 fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
875 register_subscriber_callback(&self.subscriber_callbacks, &callback);
876 if notify {
877 callback();
878 }
879 drop(callback);
880 self.subscriber_callbacks
881 .borrow_mut()
882 .retain(|callback| callback.upgrade().is_some());
883 }
884
885 #[cfg(test)]
886 fn subscriber_callback_count(&self) -> usize {
887 self.subscriber_callbacks.borrow().len()
888 }
889
890 fn notify_subscribers(&self) {
891 notify_subscriber_callbacks(&self.subscriber_callbacks);
892 }
893
894 fn notify_applied(&self) {
895 let observers = self.lock_apply_observers();
896 for observer in observers.iter() {
897 observer();
898 }
899 }
900
901 fn lock_weak_self(&self) -> MutexGuard<'_, Option<Weak<Self>>> {
902 self.weak_self
903 .lock()
904 .unwrap_or_else(|poisoned| poisoned.into_inner())
905 }
906
907 fn lock_apply_observers(&self) -> MutexGuard<'_, Vec<Box<dyn Fn() + 'static>>> {
908 self.apply_observers
909 .lock()
910 .unwrap_or_else(|poisoned| poisoned.into_inner())
911 }
912
913 fn upgrade_self(&self) -> Option<Arc<Self>> {
914 self.lock_weak_self()
915 .as_ref()
916 .and_then(|weak| weak.upgrade())
917 }
918
919 #[inline]
920 pub(crate) fn id(&self) -> ObjectId {
921 self.id
922 }
923
924 pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
925 let record = self.readable_record_for_active_snapshot().ok()?;
926 record.try_with_value(f)
927 }
928
929 pub(crate) fn try_get(&self) -> Option<T> {
930 self.try_with_value(Clone::clone)
931 }
932
933 pub(crate) fn get(&self) -> T {
934 let record = self
935 .readable_record_for_active_snapshot()
936 .unwrap_or_else(|failure| panic!("{failure}"));
937 record.with_value(|value: &T| value.clone())
938 }
939
940 pub(crate) fn set(&self, new_value: T) -> bool {
941 #[cfg(debug_assertions)]
943 {
944 let in_handler = crate::in_event_handler();
945 let in_snapshot = crate::in_applied_snapshot();
946 if in_handler && !in_snapshot {
947 log::warn!(
948 target: "cranpose::state",
949 "State modified in event handler without run_in_mutable_snapshot; \
950 this can make updates invisible to other contexts. Wrap the handler \
951 in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
952 self.id
953 );
954 }
955 }
956
957 let snapshot = active_snapshot();
958 let snapshot_id = snapshot.snapshot_id();
959
960 match &snapshot {
961 AnySnapshot::Global(global) => {
962 let invalid = snapshot.invalid();
963 let equivalent = self
964 .readable_for(snapshot_id, &invalid)
965 .map(|record| {
966 record.with_value(|current: &T| self.policy.equivalent(current, &new_value))
967 })
968 .unwrap_or(false);
969 if equivalent {
970 return false;
971 }
972
973 if global.has_pending_children() {
974 panic!(
975 "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
976 global.pending_children(),
977 self.id,
978 snapshot_id
979 );
980 }
981
982 let mut written_state: Option<Arc<dyn StateObject>> = None;
983 if let Some(state) = self.upgrade_self() {
984 let trait_object: Arc<dyn StateObject> = state.clone();
985 snapshot.record_write(trait_object.clone());
986 written_state = Some(trait_object);
987 }
988 mark_update_write(self.id);
989
990 let new_id = allocate_record_id();
991 let record = new_overwritable_record_as_head_locked(self);
992 record.replace_value(new_value);
993 record.set_snapshot_id(new_id);
994 record.set_tombstone(false);
995 advance_global_snapshot(new_id);
996 self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
997
998 if !global.has_pending_children() {
999 let mut cursor = record.next();
1000 while let Some(node) = cursor {
1001 if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
1002 node.clear_value();
1003 node.set_tombstone(true);
1004 }
1005 cursor = node.next();
1006 }
1007 self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
1008 }
1009
1010 if let Some(modified) = written_state.as_ref() {
1011 crate::snapshot_v2::notify_apply_observers(
1012 std::slice::from_ref(modified),
1013 new_id,
1014 );
1015 }
1016 }
1017 AnySnapshot::Mutable(_)
1018 | AnySnapshot::NestedMutable(_)
1019 | AnySnapshot::TransparentMutable(_) => {
1020 let invalid = snapshot.invalid();
1021 let equivalent = self
1022 .readable_for(snapshot_id, &invalid)
1023 .map(|record| {
1024 record.with_value(|current: &T| self.policy.equivalent(current, &new_value))
1025 })
1026 .unwrap_or(false);
1027 if equivalent {
1028 return false;
1029 }
1030
1031 if let Some(state) = self.upgrade_self() {
1032 let trait_object: Arc<dyn StateObject> = state.clone();
1033 snapshot.record_write(trait_object);
1034 }
1035 mark_update_write(self.id);
1036
1037 let record = self.writable_record(snapshot_id, &invalid);
1038 record.replace_value(new_value);
1039 self.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
1040 }
1041 AnySnapshot::Readonly(_)
1042 | AnySnapshot::NestedReadonly(_)
1043 | AnySnapshot::TransparentReadonly(_) => {
1044 panic!("Cannot write to a read-only snapshot");
1045 }
1046 }
1047
1048 true
1053 }
1054}
1055
1056thread_local! {
1057 static ACTIVE_UPDATES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
1058 static PENDING_WRITES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
1059}
1060
1061pub(crate) struct UpdateScope {
1062 id: ObjectId,
1063 finished: bool,
1064}
1065
1066impl UpdateScope {
1067 pub(crate) fn new(id: ObjectId) -> Self {
1068 ACTIVE_UPDATES.with(|active| {
1069 active.borrow_mut().insert(id);
1070 });
1071 PENDING_WRITES.with(|pending| {
1072 pending.borrow_mut().remove(&id);
1073 });
1074 Self {
1075 id,
1076 finished: false,
1077 }
1078 }
1079
1080 pub(crate) fn finish(mut self) -> bool {
1081 self.finished = true;
1082 ACTIVE_UPDATES.with(|active| {
1083 active.borrow_mut().remove(&self.id);
1084 });
1085 PENDING_WRITES.with(|pending| pending.borrow_mut().remove(&self.id))
1086 }
1087}
1088
1089impl Drop for UpdateScope {
1090 fn drop(&mut self) {
1091 if self.finished {
1092 return;
1093 }
1094 ACTIVE_UPDATES.with(|active| {
1095 active.borrow_mut().remove(&self.id);
1096 });
1097 PENDING_WRITES.with(|pending| {
1098 pending.borrow_mut().remove(&self.id);
1099 });
1100 }
1101}
1102
1103fn mark_update_write(id: ObjectId) {
1104 ACTIVE_UPDATES.with(|active| {
1105 if active.borrow().contains(&id) {
1106 PENDING_WRITES.with(|pending| {
1107 pending.borrow_mut().insert(id);
1108 });
1109 }
1110 });
1111}
1112
1113impl<T: Clone + 'static> SnapshotMutableState<T> {
1114 fn try_readable_record(
1116 &self,
1117 snapshot_id: SnapshotId,
1118 invalid: &SnapshotIdSet,
1119 ) -> Option<Rc<StateRecord>> {
1120 self.readable_for(snapshot_id, invalid)
1121 }
1122}
1123
1124impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
1125 fn object_id(&self) -> ObjectId {
1126 self.id
1127 }
1128
1129 fn first_record(&self) -> Rc<StateRecord> {
1130 self.head.clone_head()
1131 }
1132
1133 fn try_readable_record(
1134 &self,
1135 snapshot_id: SnapshotId,
1136 invalid: &SnapshotIdSet,
1137 ) -> Option<Rc<StateRecord>> {
1138 self.try_readable_record(snapshot_id, invalid)
1139 }
1140
1141 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
1142 self.try_readable_record(snapshot_id, invalid)
1143 .unwrap_or_else(|| {
1144 panic!(
1145 "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
1146 self.id, snapshot_id
1147 )
1148 })
1149 }
1150
1151 fn prepend_state_record(&self, record: Rc<StateRecord>) {
1152 self.head.prepend(record);
1153 }
1154
1155 fn observation_lease(&self) -> Option<Box<dyn Any>> {
1156 self.acquire_observation_lease()
1157 }
1158
1159 fn merge_records(
1160 &self,
1161 previous: Rc<StateRecord>,
1162 current: Rc<StateRecord>,
1163 applied: Rc<StateRecord>,
1164 ) -> Option<Rc<StateRecord>> {
1165 let Some(current_value) = current.try_with_value(|value: &T| value.clone()) else {
1166 log::error!(
1167 "SnapshotMutableState::merge_records current record value missing or wrong type (state {:?}, current_id={})",
1168 self.id,
1169 current.snapshot_id()
1170 );
1171 return None;
1172 };
1173 let Some(applied_value) = applied.try_with_value(|value: &T| value.clone()) else {
1174 log::error!(
1175 "SnapshotMutableState::merge_records applied record value missing or wrong type (state {:?}, applied_id={})",
1176 self.id,
1177 applied.snapshot_id()
1178 );
1179 return None;
1180 };
1181 if self.policy.equivalent(¤t_value, &applied_value) {
1182 return Some(current);
1183 }
1184
1185 let Some(previous_value) = previous.try_with_value(|value: &T| value.clone()) else {
1186 log::error!(
1187 "SnapshotMutableState::merge_records previous record value missing or wrong type (state {:?}, previous_id={})",
1188 self.id,
1189 previous.snapshot_id()
1190 );
1191 return None;
1192 };
1193 let merged = self
1194 .policy
1195 .merge(&previous_value, ¤t_value, &applied_value)?;
1196
1197 Some(StateRecord::new(applied.snapshot_id(), merged, None))
1198 }
1199
1200 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1201 let head = self.first_record();
1202 let mut cursor = Some(head);
1203 while let Some(record) = cursor {
1204 if record.snapshot_id() == child_id {
1205 let Some(cloned) = record.try_with_value(|value: &T| value.clone()) else {
1206 log::error!(
1207 "SnapshotMutableState::promote_record child record value missing or wrong type (state {:?}, child_id={})",
1208 self.id,
1209 child_id
1210 );
1211 return Err("child record value missing or wrong type");
1212 };
1213 let new_id = allocate_record_id();
1220 let promoted = new_overwritable_record_as_head_locked(self);
1221 promoted.replace_value(cloned);
1222 promoted.set_tombstone(false);
1223 promoted.set_snapshot_id(new_id);
1224 advance_global_snapshot(new_id);
1225 self.notify_applied();
1226 self.assert_chain_integrity("promote_record", Some(child_id));
1227 return Ok(());
1228 }
1229 cursor = record.next();
1230 }
1231 log::error!(
1232 "SnapshotMutableState::promote_record missing child record (state {:?}, child_id={})",
1233 self.id,
1234 child_id
1235 );
1236 Err("missing child record")
1237 }
1238
1239 fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1240 let Some(value) = merged.try_with_value(|value: &T| value.clone()) else {
1241 log::error!(
1242 "SnapshotMutableState::commit_merged_record merged record value missing or wrong type (state {:?}, merged_id={})",
1243 self.id,
1244 merged.snapshot_id()
1245 );
1246 return Err("merged record value missing or wrong type");
1247 };
1248 let new_id = allocate_record_id();
1251 let committed = new_overwritable_record_as_head_locked(self);
1252 committed.replace_value(value);
1253 committed.set_tombstone(false);
1254 committed.set_snapshot_id(new_id);
1255 advance_global_snapshot(new_id);
1256 self.notify_applied();
1257 self.assert_chain_integrity("commit_merged_record", Some(new_id));
1258 Ok(new_id)
1259 }
1260
1261 fn overwrite_unused_records(&self) -> bool {
1262 overwrite_unused_records_locked::<T>(self)
1263 }
1264
1265 fn as_any(&self) -> &dyn Any {
1266 self
1267 }
1268}
1269
1270pub(crate) struct MutableStateInner<T: Clone + 'static> {
1271 pub(crate) state: Arc<SnapshotMutableState<T>>,
1272 pub(crate) watchers: RefCell<HashMap<ScopeId, RcWeak<RecomposeScopeInner>>>,
1273 runtime: RuntimeHandle,
1274 state_id: Cell<Option<StateId>>,
1275}
1276
1277fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<RcWeak<dyn Fn()>>>) {
1278 let callbacks_snapshot = std::mem::take(&mut *callbacks.borrow_mut());
1279 let mut live = Vec::with_capacity(callbacks_snapshot.len());
1280 for callback in callbacks_snapshot {
1281 let Some(callback) = callback.upgrade() else {
1282 continue;
1283 };
1284 callback();
1285 live.push(callback);
1286 }
1287 let mut registered = callbacks.borrow_mut();
1288 registered.retain(|callback| callback.upgrade().is_some());
1289 for callback in live {
1290 let callback = Rc::downgrade(&callback);
1291 if !registered
1292 .iter()
1293 .any(|registered| registered.ptr_eq(&callback))
1294 {
1295 registered.push(callback);
1296 }
1297 }
1298}
1299
1300fn register_subscriber_callback(
1301 callbacks: &RefCell<Vec<RcWeak<dyn Fn()>>>,
1302 callback: &Rc<dyn Fn()>,
1303) {
1304 let callback_weak = Rc::downgrade(callback);
1305 let mut callbacks = callbacks.borrow_mut();
1306 callbacks.retain(|callback| callback.upgrade().is_some());
1307 if !callbacks
1308 .iter()
1309 .any(|registered| registered.ptr_eq(&callback_weak))
1310 {
1311 callbacks.push(callback_weak);
1312 }
1313}
1314
1315fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, RcWeak<RecomposeScopeInner>>) {
1316 let len = watchers.len();
1317 let capacity = watchers.capacity();
1318 if capacity > len.saturating_mul(4).max(32) {
1319 watchers.shrink_to_fit();
1320 }
1321}
1322
1323impl<T: Clone + 'static> MutableStateInner<T> {
1324 pub(crate) fn new_with_policy(
1325 value: T,
1326 runtime: RuntimeHandle,
1327 policy: Arc<dyn MutationPolicy<T>>,
1328 ) -> Self {
1329 Self {
1330 state: SnapshotMutableState::new_in_arc(value, policy),
1331 watchers: RefCell::new(HashMap::default()),
1332 runtime,
1333 state_id: Cell::new(None),
1334 }
1335 }
1336
1337 pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1338 self.state_id.set(Some(state_id));
1339 let runtime_handle = self.runtime.clone();
1340 self.state.add_apply_observer(Box::new(move || {
1341 let runtime = runtime_handle.clone();
1342 runtime_handle.enqueue_ui_task(Box::new(move || {
1343 runtime.with_state_arena(|arena| {
1344 let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1345 inner.invalidate_watchers();
1346 });
1347 });
1348 }));
1349 }));
1350 }
1351
1352 fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1353 let mut watchers = self.watchers.borrow_mut();
1354 let before = watchers.len();
1355 watchers.retain(|_, existing| existing.upgrade().is_some());
1356 self.state.remove_scope_observers(before - watchers.len());
1357 let registered = match watchers.get(&scope.id()) {
1358 Some(_) => false,
1359 _ => {
1360 watchers.insert(scope.id(), scope.downgrade());
1361 true
1362 }
1363 };
1364 drop(watchers);
1365 let became_subscribed = registered && self.state.add_scope_observer();
1366 (registered, became_subscribed)
1367 }
1368
1369 fn has_subscribers(&self) -> bool {
1370 let mut watchers = self.watchers.borrow_mut();
1371 let before = watchers.len();
1372 watchers.retain(|_, existing| existing.upgrade().is_some());
1373 self.state.remove_scope_observers(before - watchers.len());
1374 self.state.has_subscribers()
1375 }
1376
1377 pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1378 let mut watchers = self.watchers.borrow_mut();
1379 let removed = if watchers
1384 .get(&scope_id)
1385 .is_some_and(|weak| weak.upgrade().is_none())
1386 {
1387 watchers.remove(&scope_id);
1388 shrink_watchers_if_sparse(&mut watchers);
1389 true
1390 } else {
1391 false
1392 };
1393 drop(watchers);
1394 self.state.remove_scope_observers(usize::from(removed));
1395 }
1396
1397 fn state_id(&self) -> Option<StateId> {
1398 self.state_id.get()
1399 }
1400
1401 fn invalidate_watchers(&self) {
1402 let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1403 let mut watchers = self.watchers.borrow_mut();
1404 let before = watchers.len();
1405 let mut live = Vec::with_capacity(watchers.len());
1406 watchers.retain(|_, scope| {
1407 if let Some(inner) = scope.upgrade() {
1408 live.push(RecomposeScope { inner });
1409 true
1410 } else {
1411 false
1412 }
1413 });
1414 let removed_count = before - watchers.len();
1415 shrink_watchers_if_sparse(&mut watchers);
1416 (live, removed_count)
1417 };
1418 self.state.remove_scope_observers(removed_count);
1419
1420 for watcher in watchers {
1421 debug_record_scope_invalidation::<T>(watcher.id(), self.state_id.get());
1422 if let Some(state_id) = self.state_id.get() {
1423 watcher.invalidate_from_state(state_id);
1424 } else {
1425 watcher.invalidate();
1426 }
1427 }
1428 }
1429}
1430
1431impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1432 fn drop(&mut self) {
1433 self.state
1434 .remove_scope_observers(self.watchers.get_mut().len());
1435 }
1436}
1437
1438fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1439 let Some(Some(scope)) =
1440 with_current_composer_opt(|composer| composer.current_state_invalidation_scope())
1441 else {
1442 return;
1443 };
1444 let (registered, became_subscribed) = inner.register_scope(&scope);
1445 if registered {
1446 if let Some(state_id) = inner.state_id() {
1447 scope.record_state_subscription(state_id);
1448 }
1449 if became_subscribed {
1450 inner.state.notify_subscribers();
1451 }
1452 }
1453}
1454
1455pub struct State<T: Clone + 'static> {
1457 id: StateId,
1458 runtime_id: runtime::RuntimeId,
1459 _marker: PhantomData<fn() -> T>,
1460}
1461
1462pub struct MutableState<T: Clone + 'static> {
1468 id: StateId,
1469 runtime_id: runtime::RuntimeId,
1470 _marker: PhantomData<fn() -> T>,
1471}
1472
1473#[derive(Clone)]
1475pub struct OwnedMutableState<T: Clone + 'static> {
1476 state: MutableState<T>,
1477 _lease: Rc<runtime::StateHandleLease>,
1478 _marker: PhantomData<fn() -> T>,
1479}
1480
1481impl<T: Clone + 'static> PartialEq for State<T> {
1482 fn eq(&self, other: &Self) -> bool {
1483 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1484 }
1485}
1486
1487impl<T: Clone + 'static> Eq for State<T> {}
1488
1489impl<T: Clone + 'static> PartialEq for MutableState<T> {
1490 fn eq(&self, other: &Self) -> bool {
1491 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1492 }
1493}
1494
1495impl<T: Clone + 'static> Eq for MutableState<T> {}
1496
1497impl<T: Clone + 'static> Copy for State<T> {}
1498
1499impl<T: Clone + 'static> Clone for State<T> {
1500 fn clone(&self) -> Self {
1501 *self
1502 }
1503}
1504
1505impl<T: Clone + 'static> Copy for MutableState<T> {}
1506
1507impl<T: Clone + 'static> Clone for MutableState<T> {
1508 fn clone(&self) -> Self {
1509 *self
1510 }
1511}
1512
1513impl<T: Clone + 'static> State<T> {
1514 fn state_id(&self) -> StateId {
1515 self.id
1516 }
1517
1518 fn runtime_id(&self) -> runtime::RuntimeId {
1519 self.runtime_id
1520 }
1521
1522 fn runtime_handle(&self) -> RuntimeHandle {
1523 runtime::runtime_handle_by_id(self.runtime_id())
1524 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1525 }
1526
1527 fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1528 runtime::runtime_handle_by_id(self.runtime_id())
1529 }
1530
1531 fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1532 self.runtime_handle()
1533 .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1534 }
1535
1536 fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1537 self.runtime_handle_opt()?
1538 .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1539 }
1540
1541 fn subscribe_current_scope(&self) {
1542 self.with_inner(register_current_state_scope::<T>);
1543 }
1544
1545 pub fn is_alive(&self) -> bool {
1546 self.try_with_inner(|_| ()).is_some()
1547 }
1548
1549 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1550 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1551 }
1552
1553 pub fn try_value(&self) -> Option<T> {
1554 self.try_with_inner(|inner| inner.state.try_get())?
1555 }
1556
1557 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1558 let value = self.with_inner(|inner| inner.state.get());
1559 self.subscribe_current_scope();
1560 f(&value)
1561 }
1562
1563 pub fn value(&self) -> T {
1564 let value = self.with_inner(|inner| inner.state.get());
1565 self.subscribe_current_scope();
1566 value
1567 }
1568
1569 pub fn get(&self) -> T {
1570 self.value()
1571 }
1572
1573 pub fn has_subscribers(&self) -> bool {
1574 self.with_inner(MutableStateInner::has_subscribers)
1575 }
1576
1577 pub fn on_subscriber(&self, callback: Rc<dyn Fn()>) {
1578 self.with_inner(|inner| {
1579 inner
1580 .state
1581 .subscriber_callback(callback, inner.has_subscribers())
1582 });
1583 }
1584}
1585
1586impl<T: Clone + 'static> MutableState<T> {
1587 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1588 runtime.alloc_persistent_state(value)
1589 }
1590
1591 fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1592 Self {
1593 id,
1594 runtime_id,
1595 _marker: PhantomData,
1596 }
1597 }
1598
1599 pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1600 Self::from_parts(lease.id(), lease.runtime().id())
1601 }
1602
1603 fn state_id(&self) -> StateId {
1604 self.id
1605 }
1606
1607 fn runtime_id(&self) -> runtime::RuntimeId {
1608 self.runtime_id
1609 }
1610
1611 fn runtime_handle(&self) -> RuntimeHandle {
1612 runtime::runtime_handle_by_id(self.runtime_id())
1613 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1614 }
1615
1616 fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1617 runtime::runtime_handle_by_id(self.runtime_id())
1618 }
1619
1620 fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1621 self.runtime_handle()
1622 .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1623 }
1624
1625 fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1626 self.runtime_handle_opt()?
1627 .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1628 }
1629
1630 pub fn is_alive(&self) -> bool {
1631 self.try_with_inner(|_| ()).is_some()
1632 }
1633
1634 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1635 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1636 }
1637
1638 pub fn try_value(&self) -> Option<T> {
1639 self.try_with_inner(|inner| inner.state.try_get())?
1640 }
1641
1642 pub fn as_state(&self) -> State<T> {
1643 State {
1644 id: self.id,
1645 runtime_id: self.runtime_id,
1646 _marker: PhantomData,
1647 }
1648 }
1649
1650 pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1651 let lease = self
1652 .runtime_handle_opt()?
1653 .retain_state_lease(self.state_id())?;
1654 Some(OwnedMutableState {
1655 state: *self,
1656 _lease: lease,
1657 _marker: PhantomData,
1658 })
1659 }
1660
1661 pub fn retain(&self) -> OwnedMutableState<T> {
1662 self.try_retain()
1663 .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1664 }
1665
1666 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1667 let value = self.with_inner(|inner| inner.state.get());
1668 self.subscribe_current_scope();
1669 f(&value)
1670 }
1671
1672 pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1673 let runtime = self.runtime_handle();
1674 runtime.assert_ui_thread();
1675 runtime.with_state_arena(|arena| {
1676 arena.with_typed::<T, R>(self.state_id(), |inner| {
1677 let mut value = inner.state.get();
1678 let tracker = UpdateScope::new(inner.state.id());
1679 let result = f(&mut value);
1680 let wrote_elsewhere = tracker.finish();
1681 if !wrote_elsewhere && inner.state.set(value) {
1682 inner.invalidate_watchers();
1683 }
1684 result
1685 })
1686 })
1687 }
1688
1689 pub fn replace(&self, value: T) {
1690 let Some(runtime) = self.runtime_handle_opt() else {
1691 log::debug!(
1692 "MutableState::replace skipped: runtime {:?} dropped",
1693 self.runtime_id()
1694 );
1695 return;
1696 };
1697 runtime.assert_ui_thread();
1698 let replaced = runtime
1699 .try_with_state_arena(|arena| {
1700 arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1701 if inner.state.set(value) {
1702 inner.invalidate_watchers();
1703 }
1704 })
1705 })
1706 .flatten();
1707 if replaced.is_none() {
1708 log::debug!(
1709 "MutableState::replace skipped: state cell released (slot={}, gen={})",
1710 self.state_id().slot(),
1711 self.state_id().generation(),
1712 );
1713 }
1714 }
1715
1716 pub fn set_value(&self, value: T) {
1717 self.replace(value);
1718 }
1719
1720 pub fn set(&self, value: T) {
1721 self.replace(value);
1722 }
1723
1724 pub fn value(&self) -> T {
1725 let value = self.with_inner(|inner| inner.state.get());
1726 self.subscribe_current_scope();
1727 value
1728 }
1729
1730 pub fn get(&self) -> T {
1731 self.value()
1732 }
1733
1734 pub fn get_non_reactive(&self) -> T {
1735 self.with_inner(|inner| inner.state.get())
1736 }
1737
1738 #[doc(hidden)]
1739 pub fn runtime_state_id(&self) -> StateId {
1740 self.state_id()
1741 }
1742
1743 #[doc(hidden)]
1744 pub fn subscribe_current_scope_only(&self) {
1745 self.subscribe_current_scope();
1746 }
1747
1748 fn subscribe_current_scope(&self) {
1749 self.with_inner(register_current_state_scope::<T>);
1750 }
1751
1752 #[cfg(test)]
1753 pub(crate) fn watcher_count(&self) -> usize {
1754 self.with_inner(|inner| inner.watchers.borrow().len())
1755 }
1756
1757 #[cfg(test)]
1758 pub(crate) fn watcher_capacity(&self) -> usize {
1759 self.with_inner(|inner| inner.watchers.borrow().capacity())
1760 }
1761
1762 #[cfg(test)]
1763 pub(crate) fn subscriber_callback_count(&self) -> usize {
1764 self.with_inner(|inner| inner.state.subscriber_callback_count())
1765 }
1766
1767 #[cfg(test)]
1768 pub(crate) fn state_id_for_test(&self) -> StateId {
1769 self.state_id()
1770 }
1771
1772 #[cfg(test)]
1773 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1774 self.as_state().subscribe_scope_for_test(scope);
1775 }
1776}
1777
1778impl<T: Clone + 'static> OwnedMutableState<T> {
1779 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1780 let lease = runtime.alloc_state(value);
1781 Self {
1782 state: MutableState::from_lease(&lease),
1783 _lease: lease,
1784 _marker: PhantomData,
1785 }
1786 }
1787
1788 pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1789 where
1790 T: PartialEq,
1791 {
1792 Self::with_runtime_and_policy(value, runtime, Arc::new(StructuralEqual))
1793 }
1794
1795 pub(crate) fn with_runtime_and_policy(
1796 value: T,
1797 runtime: RuntimeHandle,
1798 policy: Arc<dyn MutationPolicy<T>>,
1799 ) -> Self {
1800 let lease = runtime.alloc_state_with_policy(value, policy);
1801 Self {
1802 state: MutableState::from_lease(&lease),
1803 _lease: lease,
1804 _marker: PhantomData,
1805 }
1806 }
1807
1808 pub fn handle(&self) -> MutableState<T> {
1809 self.state
1810 }
1811
1812 pub fn as_state(&self) -> State<T> {
1813 self.state.as_state()
1814 }
1815}
1816
1817impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1818 type Target = MutableState<T>;
1819
1820 fn deref(&self) -> &Self::Target {
1821 &self.state
1822 }
1823}
1824
1825#[cfg(test)]
1826impl<T: Clone + 'static> State<T> {
1827 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1828 self.with_inner(|inner| {
1829 let (registered, became_subscribed) = inner.register_scope(scope);
1830 if registered {
1831 if let Some(state_id) = inner.state_id() {
1832 scope.record_state_subscription(state_id);
1833 }
1834 if became_subscribed {
1835 inner.state.notify_subscribers();
1836 }
1837 }
1838 });
1839 }
1840}
1841
1842impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1843 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1844 if let Some(value) = self.try_value() {
1845 f.debug_struct("MutableState")
1846 .field("value", &value)
1847 .finish()
1848 } else {
1849 f.write_str("MutableState { value: <unavailable> }")
1850 }
1851 }
1852}
1853
1854#[derive(Clone)]
1855pub struct SnapshotStateList<T: Clone + 'static> {
1856 state: OwnedMutableState<Vec<T>>,
1857}
1858
1859impl<T: Clone + 'static> SnapshotStateList<T> {
1860 pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1861 where
1862 I: IntoIterator<Item = T>,
1863 {
1864 let initial: Vec<T> = values.into_iter().collect();
1865 Self {
1866 state: OwnedMutableState::with_runtime(initial, runtime),
1867 }
1868 }
1869
1870 pub fn as_state(&self) -> State<Vec<T>> {
1871 self.state.as_state()
1872 }
1873
1874 pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1875 self.state.handle()
1876 }
1877
1878 pub fn len(&self) -> usize {
1879 self.state.with(|values| values.len())
1880 }
1881
1882 pub fn is_empty(&self) -> bool {
1883 self.len() == 0
1884 }
1885
1886 pub fn to_vec(&self) -> Vec<T> {
1887 self.state.with(|values| values.clone())
1888 }
1889
1890 pub fn iter(&self) -> Vec<T> {
1891 self.to_vec()
1892 }
1893
1894 pub fn get(&self, index: usize) -> T {
1895 self.state.with(|values| values[index].clone())
1896 }
1897
1898 pub fn get_opt(&self, index: usize) -> Option<T> {
1899 self.state.with(|values| values.get(index).cloned())
1900 }
1901
1902 pub fn first(&self) -> Option<T> {
1903 self.get_opt(0)
1904 }
1905
1906 pub fn last(&self) -> Option<T> {
1907 self.state.with(|values| values.last().cloned())
1908 }
1909
1910 pub fn push(&self, value: T) {
1911 self.state.update(|values| values.push(value));
1912 }
1913
1914 pub fn extend<I>(&self, iter: I)
1915 where
1916 I: IntoIterator<Item = T>,
1917 {
1918 self.state.update(|values| values.extend(iter));
1919 }
1920
1921 pub fn insert(&self, index: usize, value: T) {
1922 self.state.update(|values| values.insert(index, value));
1923 }
1924
1925 pub fn set(&self, index: usize, value: T) -> T {
1926 self.state
1927 .update(|values| std::mem::replace(&mut values[index], value))
1928 }
1929
1930 pub fn remove(&self, index: usize) -> T {
1931 self.state.update(|values| values.remove(index))
1932 }
1933
1934 pub fn pop(&self) -> Option<T> {
1935 self.state.update(|values| values.pop())
1936 }
1937
1938 pub fn clear(&self) {
1939 self.state.replace(Vec::new());
1940 }
1941
1942 pub fn retain<F>(&self, mut predicate: F)
1943 where
1944 F: FnMut(&T) -> bool,
1945 {
1946 self.state
1947 .update(|values| values.retain(|value| predicate(value)));
1948 }
1949
1950 pub fn replace_with<I>(&self, iter: I)
1951 where
1952 I: IntoIterator<Item = T>,
1953 {
1954 self.state.replace(iter.into_iter().collect());
1955 }
1956}
1957
1958impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
1959 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1960 let contents = self.to_vec();
1961 f.debug_struct("SnapshotStateList")
1962 .field("values", &contents)
1963 .finish()
1964 }
1965}
1966
1967#[derive(Clone)]
1968pub struct SnapshotStateMap<K, V>
1969where
1970 K: Clone + Eq + Hash + 'static,
1971 V: Clone + 'static,
1972{
1973 state: OwnedMutableState<HashMap<K, V>>,
1974}
1975
1976impl<K, V> SnapshotStateMap<K, V>
1977where
1978 K: Clone + Eq + Hash + 'static,
1979 V: Clone + 'static,
1980{
1981 pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
1982 where
1983 I: IntoIterator<Item = (K, V)>,
1984 {
1985 let map: HashMap<K, V> = pairs.into_iter().collect();
1986 Self {
1987 state: OwnedMutableState::with_runtime(map, runtime),
1988 }
1989 }
1990
1991 pub fn as_state(&self) -> State<HashMap<K, V>> {
1992 self.state.as_state()
1993 }
1994
1995 pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
1996 self.state.handle()
1997 }
1998
1999 pub fn len(&self) -> usize {
2000 self.state.with(|map| map.len())
2001 }
2002
2003 pub fn is_empty(&self) -> bool {
2004 self.state.with(|map| map.is_empty())
2005 }
2006
2007 pub fn contains_key(&self, key: &K) -> bool {
2008 self.state.with(|map| map.contains_key(key))
2009 }
2010
2011 pub fn get(&self, key: &K) -> Option<V> {
2012 self.state.with(|map| map.get(key).cloned())
2013 }
2014
2015 pub fn to_hash_map(&self) -> HashMap<K, V> {
2016 self.state.with(|map| map.clone())
2017 }
2018
2019 pub fn insert(&self, key: K, value: V) -> Option<V> {
2020 self.state.update(|map| map.insert(key, value))
2021 }
2022
2023 pub fn extend<I>(&self, iter: I)
2024 where
2025 I: IntoIterator<Item = (K, V)>,
2026 {
2027 self.state.update(|map| map.extend(iter));
2028 }
2029
2030 pub fn remove(&self, key: &K) -> Option<V> {
2031 self.state.update(|map| map.remove(key))
2032 }
2033
2034 pub fn clear(&self) {
2035 self.state.replace(HashMap::default());
2036 }
2037
2038 pub fn retain<F>(&self, mut predicate: F)
2039 where
2040 F: FnMut(&K, &mut V) -> bool,
2041 {
2042 self.state.update(|map| map.retain(|k, v| predicate(k, v)));
2043 }
2044}
2045
2046impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
2047where
2048 K: Clone + Eq + Hash + fmt::Debug + 'static,
2049 V: Clone + fmt::Debug + 'static,
2050{
2051 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2052 let contents = self.to_hash_map();
2053 f.debug_struct("SnapshotStateMap")
2054 .field("entries", &contents)
2055 .finish()
2056 }
2057}
2058
2059pub(crate) struct DerivedState<T: Clone + 'static> {
2060 compute: Rc<dyn Fn() -> T>,
2061 pub(crate) state: OwnedMutableState<T>,
2062}
2063
2064impl<T: Clone + 'static> DerivedState<T> {
2065 pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
2066 let initial = compute();
2067 Self {
2068 compute,
2069 state: OwnedMutableState::with_runtime(initial, runtime),
2070 }
2071 }
2072
2073 pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
2074 self.compute = compute;
2075 }
2076
2077 pub(crate) fn recompute(&self) {
2078 let value = (self.compute)();
2079 self.state.set_value(value);
2080 }
2081}
2082
2083impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
2084 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2085 if let Some(value) = self.try_value() {
2086 f.debug_struct("State").field("value", &value).finish()
2087 } else {
2088 f.write_str("State { value: <unavailable> }")
2089 }
2090 }
2091}
2092
2093#[cfg(test)]
2094mod tests {
2095 use super::*;
2096
2097 fn create_record_chain(ids: &[SnapshotId]) -> Rc<StateRecord> {
2099 let mut head: Option<Rc<StateRecord>> = None;
2100
2101 for &id in ids.iter().rev() {
2103 head = Some(StateRecord::new(id, 0i32, head));
2104 }
2105
2106 head.expect("create_record_chain called with empty ids")
2107 }
2108
2109 struct ManualState {
2110 head: Rc<StateRecord>,
2111 }
2112
2113 impl ManualState {
2114 fn new(head: Rc<StateRecord>) -> Self {
2115 Self { head }
2116 }
2117 }
2118
2119 impl StateObject for ManualState {
2120 fn object_id(&self) -> ObjectId {
2121 ObjectId(999)
2122 }
2123
2124 fn first_record(&self) -> Rc<StateRecord> {
2125 Rc::clone(&self.head)
2126 }
2127
2128 fn try_readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Option<Rc<StateRecord>> {
2129 Some(Rc::clone(&self.head))
2130 }
2131
2132 fn readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Rc<StateRecord> {
2133 Rc::clone(&self.head)
2134 }
2135
2136 fn prepend_state_record(&self, _: Rc<StateRecord>) {}
2137
2138 fn promote_record(&self, _: SnapshotId) -> Result<(), &'static str> {
2139 Ok(())
2140 }
2141
2142 fn as_any(&self) -> &dyn Any {
2143 self
2144 }
2145 }
2146
2147 fn poison_mutex<T>(mutex: &Mutex<T>) {
2148 let poison_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2149 let _guard = mutex
2150 .lock()
2151 .unwrap_or_else(|poisoned| poisoned.into_inner());
2152 panic!("poison snapshot state mutex for recovery test");
2153 }));
2154
2155 assert!(poison_result.is_err());
2156 }
2157
2158 #[test]
2159 fn observation_leases_drive_subscriber_liveness() {
2160 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2161 let notifications = Rc::new(Cell::new(0));
2162 let notifications_for_callback = Rc::clone(¬ifications);
2163 let callback: Rc<dyn Fn()> = Rc::new(move || {
2164 notifications_for_callback.set(notifications_for_callback.get() + 1);
2165 });
2166 state.subscriber_callback(callback.clone(), false);
2167
2168 let first = StateObject::observation_lease(&*state).expect("first observation lease");
2169 let second = StateObject::observation_lease(&*state).expect("second observation lease");
2170 assert!(state.has_subscribers());
2171 assert_eq!(notifications.get(), 1);
2172
2173 drop(first);
2174 assert!(state.has_subscribers());
2175 drop(second);
2176 assert!(!state.has_subscribers());
2177
2178 let third = StateObject::observation_lease(&*state).expect("third observation lease");
2179 assert_eq!(notifications.get(), 2);
2180 drop(third);
2181 }
2182
2183 #[test]
2184 fn snapshot_mutable_state_recovers_poisoned_weak_self_lock() {
2185 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2186
2187 poison_mutex(&state.weak_self);
2188
2189 assert_eq!(state.get(), 100);
2190 assert!(state.set(101));
2191 assert_eq!(state.get(), 101);
2192 }
2193
2194 #[test]
2195 fn snapshot_mutable_state_recovers_poisoned_apply_observer_lock() {
2196 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2197 let calls = Rc::new(Cell::new(0usize));
2198 let observed_calls = Rc::clone(&calls);
2199
2200 poison_mutex(&state.apply_observers);
2201
2202 state.add_apply_observer(Box::new(move || {
2203 observed_calls.set(observed_calls.get() + 1);
2204 }));
2205 state.notify_applied();
2206
2207 assert_eq!(calls.get(), 1);
2208 }
2209
2210 #[test]
2211 fn snapshot_mutable_state_promote_missing_record_returns_error() {
2212 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2213 let missing_snapshot = usize::MAX - 17;
2214
2215 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2216 StateObject::promote_record(&*state, missing_snapshot)
2217 }));
2218
2219 assert!(
2220 matches!(result, Ok(Err("missing child record"))),
2221 "missing child record should be reported through Result, got {result:?}"
2222 );
2223 }
2224
2225 #[test]
2226 fn snapshot_mutable_state_promote_wrong_record_type_returns_error() {
2227 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2228 let child_snapshot = usize::MAX - 31;
2229 let wrong_record = StateRecord::new(child_snapshot, "wrong type", None);
2230 StateObject::prepend_state_record(&*state, wrong_record);
2231
2232 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2233 StateObject::promote_record(&*state, child_snapshot)
2234 }));
2235
2236 assert!(
2237 matches!(result, Ok(Err("child record value missing or wrong type"))),
2238 "wrong child record type should be reported through Result, got {result:?}"
2239 );
2240 }
2241
2242 #[test]
2243 fn snapshot_mutable_state_commit_wrong_record_type_returns_error() {
2244 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2245 let merged = StateRecord::new(usize::MAX - 43, "wrong type", None);
2246
2247 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2248 StateObject::commit_merged_record(&*state, merged)
2249 }));
2250
2251 assert!(
2252 matches!(result, Ok(Err("merged record value missing or wrong type"))),
2253 "wrong merged record type should be reported through Result, got {result:?}"
2254 );
2255 }
2256
2257 #[test]
2258 fn snapshot_mutable_state_merge_wrong_record_type_returns_none() {
2259 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2260 let previous = StateRecord::new(usize::MAX - 51, 1i32, None);
2261 let current = StateRecord::new(usize::MAX - 52, "wrong type", None);
2262 let applied = StateRecord::new(usize::MAX - 53, 2i32, None);
2263
2264 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2265 StateObject::merge_records(&*state, previous, current, applied)
2266 }));
2267
2268 match result {
2269 Ok(None) => {}
2270 Ok(Some(_)) => panic!("wrong merge record type unexpectedly produced a merged record"),
2271 Err(_) => panic!("wrong merge record type should not panic"),
2272 }
2273 }
2274
2275 #[test]
2276 fn test_used_locked_finds_invalid_snapshot() {
2277 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2279 let invalid_rec = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, Some(tail));
2280 let head = StateRecord::new(10, 0i32, Some(invalid_rec.clone()));
2281
2282 let result = used_locked(&head);
2283 assert!(result.is_some());
2284 assert_eq!(result.unwrap().snapshot_id(), INVALID_SNAPSHOT_ID);
2285 }
2286
2287 #[test]
2288 fn test_used_locked_finds_obscured_record() {
2289 crate::snapshot_pinning::reset_pinning_table();
2291
2292 let pin_handle = crate::snapshot_pinning::track_pinning(10, &SnapshotIdSet::EMPTY);
2295
2296 let oldest = StateRecord::new(2, 0i32, None);
2298 let newer = StateRecord::new(5, 0i32, Some(oldest.clone()));
2299 let head = StateRecord::new(100, 0i32, Some(newer));
2300
2301 let result = used_locked(&head);
2302
2303 assert!(result.is_some());
2305 let reused = result.unwrap();
2306 assert_eq!(
2307 reused.snapshot_id(),
2308 2,
2309 "Should return the oldest obscured record"
2310 );
2311
2312 crate::snapshot_pinning::release_pinning(pin_handle);
2314 }
2315
2316 #[test]
2317 fn test_used_locked_no_reusable_record() {
2318 crate::snapshot_pinning::reset_pinning_table();
2320
2321 let high_id = allocate_record_id() + 1000;
2324 let head = create_record_chain(&[high_id, high_id + 1, high_id + 2]);
2325
2326 let result = used_locked(&head);
2327 assert!(
2328 result.is_none(),
2329 "Should find no reusable records when all are recent"
2330 );
2331 }
2332
2333 #[test]
2334 fn test_used_locked_single_old_record() {
2335 crate::snapshot_pinning::reset_pinning_table();
2337
2338 let old = StateRecord::new(2, 0i32, None);
2340 let head = StateRecord::new(100, 0i32, Some(old));
2341
2342 let result = used_locked(&head);
2343 assert!(result.is_none(), "Single old record should not be reused");
2345 }
2346
2347 #[test]
2348 fn test_readable_record_for_preexisting() {
2349 let head = create_record_chain(&[PREEXISTING_SNAPSHOT_ID]);
2350 let invalid = SnapshotIdSet::EMPTY;
2351
2352 let result = readable_record_for(&head, 10, &invalid);
2353 assert!(result.is_some());
2354 assert_eq!(result.unwrap().snapshot_id(), PREEXISTING_SNAPSHOT_ID);
2355 }
2356
2357 #[test]
2358 fn test_readable_record_for_picks_highest_valid() {
2359 let head = create_record_chain(&[10, 5, PREEXISTING_SNAPSHOT_ID]);
2360 let invalid = SnapshotIdSet::EMPTY;
2361
2362 let result = readable_record_for(&head, 10, &invalid);
2364 assert!(result.is_some());
2365 assert_eq!(result.unwrap().snapshot_id(), 10);
2366
2367 let result = readable_record_for(&head, 7, &invalid);
2369 assert!(result.is_some());
2370 assert_eq!(result.unwrap().snapshot_id(), 5);
2371 }
2372
2373 #[test]
2374 fn test_new_overwritable_record_locked_reuses_invalid() {
2375 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2377
2378 let current_head = state.first_record();
2380 let invalid_rec = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, current_head.next());
2381 current_head.set_next(Some(invalid_rec.clone()));
2382
2383 let result = new_overwritable_record_locked(&*state);
2384
2385 assert!(Rc::ptr_eq(&result, &invalid_rec));
2387 assert_eq!(result.snapshot_id(), SNAPSHOT_ID_MAX);
2388 }
2389
2390 #[test]
2391 fn test_new_overwritable_record_locked_creates_new() {
2392 crate::snapshot_pinning::reset_pinning_table();
2393
2394 let _pin_handle = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2397
2398 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2400 let old_head = state.first_record();
2401
2402 let result = new_overwritable_record_locked(&*state);
2403
2404 assert_eq!(result.snapshot_id(), SNAPSHOT_ID_MAX);
2406
2407 let new_head = state.first_record();
2409 assert!(
2410 Rc::ptr_eq(&new_head, &result),
2411 "new_head ({:p}) should equal result ({:p})",
2412 Rc::as_ptr(&new_head),
2413 Rc::as_ptr(&result)
2414 );
2415
2416 assert!(result.next().is_some());
2418 assert!(Rc::ptr_eq(&result.next().unwrap(), &old_head));
2419 }
2420
2421 #[test]
2422 fn test_writable_record_reuses_invalid_record() {
2423 crate::snapshot_pinning::reset_pinning_table();
2424
2425 let state = SnapshotMutableState::new_in_arc(7i32, Arc::new(NeverEqual));
2426
2427 let head = state.first_record();
2429 let invalid = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, head.next());
2430 head.set_next(Some(invalid.clone()));
2431
2432 let snapshot_id = allocate_record_id();
2433 let result = state.writable_record(snapshot_id, &SnapshotIdSet::EMPTY);
2434
2435 assert!(
2436 Rc::ptr_eq(&result, &invalid),
2437 "Expected writable_record to reuse the INVALID record"
2438 );
2439 assert_eq!(result.snapshot_id(), snapshot_id);
2440 result.with_value(|value: &i32| {
2441 assert_eq!(*value, 7, "Reused record should copy the readable value");
2442 });
2443 assert!(!result.is_tombstone());
2444 }
2445
2446 #[test]
2447 fn test_writable_record_creates_new_when_reuse_disallowed() {
2448 crate::snapshot_pinning::reset_pinning_table();
2449 let pin = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2450
2451 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2452 let original_head = state.first_record();
2453 let preexisting = original_head
2454 .next()
2455 .expect("preexisting record should exist for newly created state");
2456
2457 let snapshot_id = allocate_record_id();
2458 let result = state.writable_record(snapshot_id, &SnapshotIdSet::EMPTY);
2459
2460 assert!(
2461 !Rc::ptr_eq(&result, &original_head),
2462 "Should not reuse the current head when reuse is disallowed"
2463 );
2464 assert!(
2465 !Rc::ptr_eq(&result, &preexisting),
2466 "Should not reuse the PREEXISTING record"
2467 );
2468 assert_eq!(result.snapshot_id(), snapshot_id);
2469 result.with_value(|value: &i32| assert_eq!(*value, 42));
2470
2471 let new_head = state.first_record();
2472 assert!(
2473 Rc::ptr_eq(&new_head, &result),
2474 "Newly created record should become the head of the chain"
2475 );
2476
2477 crate::snapshot_pinning::release_pinning(pin);
2478 }
2479
2480 #[test]
2481 fn test_state_record_clear_for_reuse() {
2482 let record = StateRecord::new(10, 42i32, None);
2483
2484 record.with_value(|val: &i32| {
2486 assert_eq!(*val, 42);
2487 });
2488
2489 record.clear_for_reuse();
2491
2492 assert_eq!(record.snapshot_id(), 10);
2495 }
2496
2497 #[test]
2498 fn test_overwrite_unused_records_no_old_records() {
2499 crate::snapshot_pinning::reset_pinning_table();
2500
2501 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2503
2504 let _pin = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2507
2508 let should_retain = state.overwrite_unused_records();
2509
2510 assert!(
2512 should_retain,
2513 "Should retain multiple records when none are old enough"
2514 );
2515
2516 let mut cursor = Some(state.first_record());
2518 while let Some(record) = cursor {
2519 assert_ne!(record.snapshot_id(), INVALID_SNAPSHOT_ID);
2520 cursor = record.next();
2521 }
2522 }
2523
2524 #[test]
2525 fn test_overwrite_unused_records_basic_cleanup() {
2526 crate::snapshot_pinning::reset_pinning_table();
2528
2529 let rec1 = StateRecord::new(100, 1i32, None);
2531 let rec2 = StateRecord::new(200, 2i32, Some(rec1.clone()));
2532 let rec3 = StateRecord::new(300, 3i32, Some(rec2.clone()));
2533
2534 struct TestState {
2536 head: Rc<StateRecord>,
2537 }
2538 impl StateObject for TestState {
2539 fn object_id(&self) -> ObjectId {
2540 ObjectId(999)
2541 }
2542 fn first_record(&self) -> Rc<StateRecord> {
2543 Rc::clone(&self.head)
2544 }
2545 fn try_readable_record(
2546 &self,
2547 _: SnapshotId,
2548 _: &SnapshotIdSet,
2549 ) -> Option<Rc<StateRecord>> {
2550 Some(Rc::clone(&self.head))
2551 }
2552 fn readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Rc<StateRecord> {
2553 Rc::clone(&self.head)
2554 }
2555 fn prepend_state_record(&self, _: Rc<StateRecord>) {}
2556 fn promote_record(&self, _: SnapshotId) -> Result<(), &'static str> {
2557 Ok(())
2558 }
2559 fn as_any(&self) -> &dyn Any {
2560 self
2561 }
2562 }
2563
2564 let test_state = TestState { head: rec3.clone() };
2565
2566 let _pin = crate::snapshot_pinning::track_pinning(1000, &SnapshotIdSet::EMPTY);
2568
2569 let result = overwrite_unused_records_locked::<i32>(&test_state);
2570
2571 assert_eq!(rec3.snapshot_id(), 300);
2573 assert_eq!(rec2.snapshot_id(), INVALID_SNAPSHOT_ID);
2574 assert_eq!(rec1.snapshot_id(), INVALID_SNAPSHOT_ID);
2575
2576 assert!(!result);
2578 }
2579
2580 #[test]
2581 fn test_overwrite_unused_records_single_record_only() {
2582 crate::snapshot_pinning::reset_pinning_table();
2583
2584 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2585
2586 let head = state.first_record();
2588 head.set_next(None);
2589
2590 let should_retain = state.overwrite_unused_records();
2591
2592 assert!(!should_retain, "Single record should return false");
2594 }
2595
2596 #[test]
2597 fn snapshot_state_try_get_reports_missing_visible_record_without_panicking() {
2598 crate::snapshot_pinning::reset_pinning_table();
2599
2600 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2601 let head = state.first_record();
2602 head.set_snapshot_id(SNAPSHOT_ID_MAX);
2603 head.set_next(None);
2604
2605 assert_eq!(state.try_get(), None);
2606 }
2607
2608 #[test]
2609 fn test_overwrite_unused_records_clears_values() {
2610 crate::snapshot_pinning::reset_pinning_table();
2611
2612 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2613 let old_rec1 = StateRecord::new(2, 999i32, Some(tail.clone()));
2614 let old_rec2 = StateRecord::new(3, 888i32, Some(old_rec1.clone()));
2615 let head = StateRecord::new(150, 42i32, Some(old_rec2.clone()));
2616 let state = ManualState::new(head.clone());
2617
2618 old_rec1.with_value(|val: &i32| {
2620 assert_eq!(*val, 999);
2621 });
2622
2623 let _pin = crate::snapshot_pinning::track_pinning(100, &SnapshotIdSet::EMPTY);
2624 overwrite_unused_records_locked::<i32>(&state);
2625
2626 assert_eq!(old_rec1.snapshot_id(), INVALID_SNAPSHOT_ID);
2628 }
2630
2631 #[test]
2632 fn test_overwrite_unused_records_mixed_old_and_new() {
2633 crate::snapshot_pinning::reset_pinning_table();
2634
2635 let preexisting = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2637 let rec2 = StateRecord::new(2, 100i32, Some(preexisting.clone()));
2638 let rec5 = StateRecord::new(5, 100i32, Some(rec2.clone()));
2639 let rec50 = StateRecord::new(50, 100i32, Some(rec5.clone()));
2640 let head = StateRecord::new(120, 100i32, Some(rec50.clone()));
2641 let state = ManualState::new(head.clone());
2642
2643 let _pin = crate::snapshot_pinning::track_pinning(40, &SnapshotIdSet::EMPTY);
2645
2646 let should_retain = overwrite_unused_records_locked::<i32>(&state);
2647 assert!(should_retain);
2648
2649 assert_eq!(rec50.snapshot_id(), 50);
2651 assert_eq!(rec5.snapshot_id(), 5);
2653 assert_eq!(rec2.snapshot_id(), INVALID_SNAPSHOT_ID);
2655 }
2656
2657 #[test]
2658 fn test_readable_record_for_skips_invalid_set() {
2659 let head = create_record_chain(&[10, 5, PREEXISTING_SNAPSHOT_ID]);
2660 let invalid = SnapshotIdSet::new().set(5);
2661
2662 let result = readable_record_for(&head, 10, &invalid);
2664 assert!(result.is_some());
2665 assert_eq!(result.unwrap().snapshot_id(), 10);
2666
2667 let result = readable_record_for(&head, 7, &invalid);
2669 assert!(result.is_some());
2670 assert_eq!(result.unwrap().snapshot_id(), PREEXISTING_SNAPSHOT_ID);
2671 }
2672
2673 #[test]
2676 fn test_assign_value_copies_int() {
2677 let source = StateRecord::new(10, 42i32, None);
2678 let target = StateRecord::new(20, 0i32, None);
2679
2680 target.assign_value::<i32>(&source).expect("copy int value");
2681
2682 target.with_value(|val: &i32| {
2684 assert_eq!(*val, 42);
2685 });
2686
2687 source.with_value(|val: &i32| {
2689 assert_eq!(*val, 42);
2690 });
2691
2692 assert_eq!(source.snapshot_id(), 10);
2694 assert_eq!(target.snapshot_id(), 20);
2695 }
2696
2697 #[test]
2698 fn test_assign_value_copies_string() {
2699 let source = StateRecord::new(10, "hello".to_string(), None);
2700 let target = StateRecord::new(20, "world".to_string(), None);
2701
2702 target
2703 .assign_value::<String>(&source)
2704 .expect("copy string value");
2705
2706 target.with_value(|val: &String| {
2708 assert_eq!(val, "hello");
2709 });
2710
2711 source.with_value(|val: &String| {
2713 assert_eq!(val, "hello");
2714 });
2715 }
2716
2717 #[test]
2718 fn test_assign_value_reports_cleared_source() {
2719 let source = StateRecord::new(10, 42i32, None);
2720 let target = StateRecord::new(20, 0i32, None);
2721
2722 source.clear_value();
2723
2724 assert_eq!(
2725 target.assign_value::<i32>(&source),
2726 Err(StateRecordValueError::MissingOrWrongType {
2727 expected: std::any::type_name::<i32>(),
2728 })
2729 );
2730 assert_eq!(target.with_value(|val: &i32| *val), 0);
2731 }
2732
2733 #[test]
2734 fn test_assign_value_overwrites_existing_value() {
2735 let source = StateRecord::new(10, 100i32, None);
2736 let target = StateRecord::new(20, 999i32, None);
2737
2738 target.with_value(|val: &i32| {
2740 assert_eq!(*val, 999);
2741 });
2742
2743 target
2745 .assign_value::<i32>(&source)
2746 .expect("overwrite int value");
2747
2748 target.with_value(|val: &i32| {
2750 assert_eq!(*val, 100);
2751 });
2752 }
2753
2754 #[test]
2755 fn test_assign_value_with_custom_type() {
2756 #[derive(Clone, PartialEq, Debug)]
2757 struct Point {
2758 x: f64,
2759 y: f64,
2760 }
2761
2762 let source = StateRecord::new(10, Point { x: 1.5, y: 2.5 }, None);
2763 let target = StateRecord::new(20, Point { x: 0.0, y: 0.0 }, None);
2764
2765 target
2766 .assign_value::<Point>(&source)
2767 .expect("copy point value");
2768
2769 target.with_value(|val: &Point| {
2770 assert_eq!(val, &Point { x: 1.5, y: 2.5 });
2771 });
2772 }
2773
2774 #[test]
2775 fn test_assign_value_self_assignment() {
2776 let record = StateRecord::new(10, 42i32, None);
2777
2778 record
2780 .assign_value::<i32>(&record)
2781 .expect("self-assign int value");
2782
2783 record.with_value(|val: &i32| {
2784 assert_eq!(*val, 42);
2785 });
2786 }
2787
2788 #[test]
2795 fn event_loop_writes_keep_the_record_chain_bounded() {
2796 crate::snapshot_pinning::reset_pinning_table();
2797 let state = SnapshotMutableState::new_in_arc(0.0f32, Arc::new(NeverEqual));
2798
2799 let mut lens = Vec::new();
2800 for event in 0..3000usize {
2801 crate::run_in_mutable_snapshot(|| {
2802 state.set(event as f32);
2803 })
2804 .expect("event snapshot applies");
2805 let _ = state.get();
2808 if event % 500 == 499 {
2809 lens.push(state.record_chain_debug().len());
2810 }
2811 }
2812
2813 let final_len = *lens.last().expect("sampled chain lengths");
2814 assert!(
2815 final_len <= 16,
2816 "record chain grew without bound across event-loop writes: {lens:?}"
2817 );
2818 }
2819
2820 #[test]
2821 fn test_assign_value_with_vec() {
2822 let source = StateRecord::new(10, vec![1, 2, 3, 4, 5], None);
2823 let target = StateRecord::new(20, Vec::<i32>::new(), None);
2824
2825 target
2826 .assign_value::<Vec<i32>>(&source)
2827 .expect("copy vec value");
2828
2829 target.with_value(|val: &Vec<i32>| {
2830 assert_eq!(val, &vec![1, 2, 3, 4, 5]);
2831 });
2832
2833 source.replace_value(vec![10, 20]);
2835 target.with_value(|val: &Vec<i32>| {
2836 assert_eq!(val, &vec![1, 2, 3, 4, 5]);
2837 });
2838 }
2839}