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