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