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