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 is_equivalent_to_readable(
631 &self,
632 snapshot_id: SnapshotId,
633 invalid: &SnapshotIdSet,
634 new_value: &T,
635 ) -> bool {
636 self.readable_for(snapshot_id, invalid)
637 .map(|record| {
638 record.with_value(|current: &T| self.policy.equivalent(current, new_value))
639 })
640 .unwrap_or(false)
641 }
642
643 fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
644 let readable = match self.readable_for(snapshot_id, invalid) {
645 Some(record) => record,
646 None => {
647 let current_head = self.head.clone_head();
648 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid);
649 let source = refreshed.unwrap_or_else(|| current_head.clone());
650
651 let cloned_value = source.with_value(|value: &T| value.clone());
652 let new_head = StateRecord::new(snapshot_id, cloned_value, Some(current_head));
653 self.head.replace(new_head.clone());
654 self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
655 return new_head;
656 }
657 };
658
659 if readable.snapshot_id() == snapshot_id {
660 return readable;
661 }
662
663 let refreshed = {
664 let current_head = self.head.clone_head();
665 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid).unwrap_or_else(
666 || {
667 panic!(
668 "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
669 self.id, snapshot_id, invalid
670 )
671 },
672 );
673
674 if refreshed.snapshot_id() == snapshot_id {
675 return refreshed;
676 }
677
678 Rc::clone(&refreshed)
679 };
680
681 let overwritable = new_overwritable_record_locked(self);
682 if let Err(error) = overwritable.assign_value::<T>(&refreshed) {
683 log::error!(
684 "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
685 self.id,
686 error
687 );
688 }
689 overwritable.set_snapshot_id(snapshot_id);
690 overwritable.set_tombstone(false);
691
692 self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
693
694 overwritable
695 }
696
697 pub(crate) fn new_in_arc(initial: T, policy: Arc<dyn MutationPolicy<T>>) -> Arc<Self> {
698 let snapshot = active_snapshot();
699 let snapshot_id = snapshot.snapshot_id();
700
701 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
702 let head = StateRecord::new(snapshot_id, initial, Some(tail));
703
704 let mut state = Arc::new(Self {
705 head: CurrentRecord::new(head),
706 policy,
707 id: ObjectId::default(),
708 weak_self: Mutex::new(None),
709 apply_observers: Mutex::new(Vec::new()),
710 read_observation_count: Cell::new(0),
711 scope_observation_count: Cell::new(0),
712 subscriber_callbacks: RefCell::new(Vec::new()),
713 });
714
715 let id = ObjectId::new(&state);
716 if let Some(state_inner) = Arc::get_mut(&mut state) {
717 state_inner.id = id;
718 }
719
720 *state.lock_weak_self() = Some(Arc::downgrade(&state));
721
722 state
723 }
724
725 pub(crate) fn add_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
726 self.lock_apply_observers().push(observer);
727 }
728
729 fn acquire_observation_lease(&self) -> Option<Box<dyn Any>> {
730 let state = self.lock_weak_self().as_ref()?.clone();
731 let was_empty = !self.has_subscribers();
732 let count = self
733 .read_observation_count
734 .get()
735 .checked_add(1)
736 .expect("state observation lease count overflow");
737 self.read_observation_count.set(count);
738 if was_empty {
739 notify_subscriber_callbacks(&self.subscriber_callbacks);
740 }
741 Some(Box::new(StateObservationLease { state }))
742 }
743
744 fn add_scope_observer(&self) -> bool {
745 let was_empty = !self.has_subscribers();
746 let count = self
747 .scope_observation_count
748 .get()
749 .checked_add(1)
750 .expect("state scope observation count overflow");
751 self.scope_observation_count.set(count);
752 was_empty
753 }
754
755 fn remove_scope_observers(&self, count: usize) {
756 if count == 0 {
757 return;
758 }
759 let remaining = self
760 .scope_observation_count
761 .get()
762 .checked_sub(count)
763 .expect("state scope observation count underflow");
764 self.scope_observation_count.set(remaining);
765 }
766
767 fn has_subscribers(&self) -> bool {
768 self.read_observation_count.get() > 0 || self.scope_observation_count.get() > 0
769 }
770
771 fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
772 register_subscriber_callback(&self.subscriber_callbacks, &callback);
773 if notify {
774 callback();
775 }
776 drop(callback);
777 self.subscriber_callbacks
778 .borrow_mut()
779 .retain(|callback| callback.upgrade().is_some());
780 }
781
782 #[cfg(test)]
783 fn subscriber_callback_count(&self) -> usize {
784 self.subscriber_callbacks.borrow().len()
785 }
786
787 fn notify_subscribers(&self) {
788 notify_subscriber_callbacks(&self.subscriber_callbacks);
789 }
790
791 fn notify_applied(&self) {
792 let observers = self.lock_apply_observers();
793 for observer in observers.iter() {
794 observer();
795 }
796 }
797
798 fn lock_weak_self(&self) -> MutexGuard<'_, Option<Weak<Self>>> {
799 self.weak_self
800 .lock()
801 .unwrap_or_else(|poisoned| poisoned.into_inner())
802 }
803
804 fn lock_apply_observers(&self) -> MutexGuard<'_, Vec<Box<dyn Fn() + 'static>>> {
805 self.apply_observers
806 .lock()
807 .unwrap_or_else(|poisoned| poisoned.into_inner())
808 }
809
810 fn upgrade_self(&self) -> Option<Arc<Self>> {
811 self.lock_weak_self()
812 .as_ref()
813 .and_then(|weak| weak.upgrade())
814 }
815
816 #[inline]
817 pub(crate) fn id(&self) -> ObjectId {
818 self.id
819 }
820
821 pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
822 let record = self.readable_record_for_active_snapshot().ok()?;
823 record.try_with_value(f)
824 }
825
826 pub(crate) fn try_get(&self) -> Option<T> {
827 self.try_with_value(Clone::clone)
828 }
829
830 pub(crate) fn get(&self) -> T {
831 let record = self
832 .readable_record_for_active_snapshot()
833 .unwrap_or_else(|failure| panic!("{failure}"));
834 record.with_value(|value: &T| value.clone())
835 }
836
837 pub(crate) fn set(&self, new_value: T) -> bool {
838 #[cfg(debug_assertions)]
839 {
840 let in_handler = crate::in_event_handler();
841 let in_snapshot = crate::in_applied_snapshot();
842 if in_handler && !in_snapshot {
843 log::warn!(
844 target: "cranpose::state",
845 "State modified in event handler without run_in_mutable_snapshot; \
846 this can make updates invisible to other contexts. Wrap the handler \
847 in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
848 self.id
849 );
850 }
851 }
852
853 let snapshot = active_snapshot();
854 let snapshot_id = snapshot.snapshot_id();
855
856 match &snapshot {
857 AnySnapshot::Global(global) => {
858 let invalid = snapshot.invalid();
859 if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
860 return false;
861 }
862
863 if global.has_pending_children() {
864 panic!(
865 "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
866 global.pending_children(),
867 self.id,
868 snapshot_id
869 );
870 }
871
872 let mut written_state: Option<Arc<dyn StateObject>> = None;
873 if let Some(state) = self.upgrade_self() {
874 let trait_object: Arc<dyn StateObject> = state.clone();
875 snapshot.record_write(trait_object.clone());
876 written_state = Some(trait_object);
877 }
878 mark_update_write(self.id);
879
880 let new_id = allocate_record_id();
881 let record = new_overwritable_record_as_head_locked(self);
882 record.replace_value(new_value);
883 record.set_snapshot_id(new_id);
884 record.set_tombstone(false);
885 advance_global_snapshot(new_id);
886 self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
887
888 if !global.has_pending_children() {
889 let mut cursor = record.next();
890 while let Some(node) = cursor {
891 if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
892 node.clear_value();
893 node.set_tombstone(true);
894 }
895 cursor = node.next();
896 }
897 self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
898 }
899
900 if let Some(modified) = written_state.as_ref() {
901 crate::snapshot_v2::notify_apply_observers(
902 std::slice::from_ref(modified),
903 new_id,
904 );
905 }
906 }
907 AnySnapshot::Mutable(_)
908 | AnySnapshot::NestedMutable(_)
909 | AnySnapshot::TransparentMutable(_) => {
910 let invalid = snapshot.invalid();
911 if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
912 return false;
913 }
914
915 if let Some(state) = self.upgrade_self() {
916 let trait_object: Arc<dyn StateObject> = state.clone();
917 snapshot.record_write(trait_object);
918 }
919 mark_update_write(self.id);
920
921 let record = self.writable_record(snapshot_id, &invalid);
922 record.replace_value(new_value);
923 self.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
924 }
925 AnySnapshot::Readonly(_)
926 | AnySnapshot::NestedReadonly(_)
927 | AnySnapshot::TransparentReadonly(_) => {
928 panic!("Cannot write to a read-only snapshot");
929 }
930 }
931
932 true
933 }
934}
935
936thread_local! {
937 static ACTIVE_UPDATES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
938 static PENDING_WRITES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
939}
940
941pub(crate) struct UpdateScope {
942 id: ObjectId,
943 finished: bool,
944}
945
946impl UpdateScope {
947 pub(crate) fn new(id: ObjectId) -> Self {
948 ACTIVE_UPDATES.with(|active| {
949 active.borrow_mut().insert(id);
950 });
951 PENDING_WRITES.with(|pending| {
952 pending.borrow_mut().remove(&id);
953 });
954 Self {
955 id,
956 finished: false,
957 }
958 }
959
960 pub(crate) fn finish(mut self) -> bool {
961 self.finished = true;
962 ACTIVE_UPDATES.with(|active| {
963 active.borrow_mut().remove(&self.id);
964 });
965 PENDING_WRITES.with(|pending| pending.borrow_mut().remove(&self.id))
966 }
967}
968
969impl Drop for UpdateScope {
970 fn drop(&mut self) {
971 if self.finished {
972 return;
973 }
974 ACTIVE_UPDATES.with(|active| {
975 active.borrow_mut().remove(&self.id);
976 });
977 PENDING_WRITES.with(|pending| {
978 pending.borrow_mut().remove(&self.id);
979 });
980 }
981}
982
983fn mark_update_write(id: ObjectId) {
984 ACTIVE_UPDATES.with(|active| {
985 if active.borrow().contains(&id) {
986 PENDING_WRITES.with(|pending| {
987 pending.borrow_mut().insert(id);
988 });
989 }
990 });
991}
992
993impl<T: Clone + 'static> SnapshotMutableState<T> {
994 fn try_readable_record(
995 &self,
996 snapshot_id: SnapshotId,
997 invalid: &SnapshotIdSet,
998 ) -> Option<Rc<StateRecord>> {
999 self.readable_for(snapshot_id, invalid)
1000 }
1001}
1002
1003impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
1004 fn object_id(&self) -> ObjectId {
1005 self.id
1006 }
1007
1008 fn first_record(&self) -> Rc<StateRecord> {
1009 self.head.clone_head()
1010 }
1011
1012 fn try_readable_record(
1013 &self,
1014 snapshot_id: SnapshotId,
1015 invalid: &SnapshotIdSet,
1016 ) -> Option<Rc<StateRecord>> {
1017 self.try_readable_record(snapshot_id, invalid)
1018 }
1019
1020 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
1021 self.try_readable_record(snapshot_id, invalid)
1022 .unwrap_or_else(|| {
1023 panic!(
1024 "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
1025 self.id, snapshot_id
1026 )
1027 })
1028 }
1029
1030 fn prepend_state_record(&self, record: Rc<StateRecord>) {
1031 self.head.prepend(record);
1032 }
1033
1034 fn observation_lease(&self) -> Option<Box<dyn Any>> {
1035 self.acquire_observation_lease()
1036 }
1037
1038 fn merge_records(
1039 &self,
1040 previous: Rc<StateRecord>,
1041 current: Rc<StateRecord>,
1042 applied: Rc<StateRecord>,
1043 ) -> Option<Rc<StateRecord>> {
1044 let Some(current_value) = current.try_with_value(|value: &T| value.clone()) else {
1045 log::error!(
1046 "SnapshotMutableState::merge_records current record value missing or wrong type (state {:?}, current_id={})",
1047 self.id,
1048 current.snapshot_id()
1049 );
1050 return None;
1051 };
1052 let Some(applied_value) = applied.try_with_value(|value: &T| value.clone()) else {
1053 log::error!(
1054 "SnapshotMutableState::merge_records applied record value missing or wrong type (state {:?}, applied_id={})",
1055 self.id,
1056 applied.snapshot_id()
1057 );
1058 return None;
1059 };
1060 if self.policy.equivalent(¤t_value, &applied_value) {
1061 return Some(current);
1062 }
1063
1064 let Some(previous_value) = previous.try_with_value(|value: &T| value.clone()) else {
1065 log::error!(
1066 "SnapshotMutableState::merge_records previous record value missing or wrong type (state {:?}, previous_id={})",
1067 self.id,
1068 previous.snapshot_id()
1069 );
1070 return None;
1071 };
1072 let merged = self
1073 .policy
1074 .merge(&previous_value, ¤t_value, &applied_value)?;
1075
1076 Some(StateRecord::new(applied.snapshot_id(), merged, None))
1077 }
1078
1079 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1080 let head = self.first_record();
1081 let mut cursor = Some(head);
1082 while let Some(record) = cursor {
1083 if record.snapshot_id() == child_id {
1084 let Some(cloned) = record.try_with_value(|value: &T| value.clone()) else {
1085 log::error!(
1086 "SnapshotMutableState::promote_record child record value missing or wrong type (state {:?}, child_id={})",
1087 self.id,
1088 child_id
1089 );
1090 return Err("child record value missing or wrong type");
1091 };
1092 let new_id = allocate_record_id();
1093 let promoted = new_overwritable_record_as_head_locked(self);
1094 promoted.replace_value(cloned);
1095 promoted.set_tombstone(false);
1096 promoted.set_snapshot_id(new_id);
1097 advance_global_snapshot(new_id);
1098 self.notify_applied();
1099 self.assert_chain_integrity("promote_record", Some(child_id));
1100 return Ok(());
1101 }
1102 cursor = record.next();
1103 }
1104 log::error!(
1105 "SnapshotMutableState::promote_record missing child record (state {:?}, child_id={})",
1106 self.id,
1107 child_id
1108 );
1109 Err("missing child record")
1110 }
1111
1112 fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1113 let Some(value) = merged.try_with_value(|value: &T| value.clone()) else {
1114 log::error!(
1115 "SnapshotMutableState::commit_merged_record merged record value missing or wrong type (state {:?}, merged_id={})",
1116 self.id,
1117 merged.snapshot_id()
1118 );
1119 return Err("merged record value missing or wrong type");
1120 };
1121 let new_id = allocate_record_id();
1122 let committed = new_overwritable_record_as_head_locked(self);
1123 committed.replace_value(value);
1124 committed.set_tombstone(false);
1125 committed.set_snapshot_id(new_id);
1126 advance_global_snapshot(new_id);
1127 self.notify_applied();
1128 self.assert_chain_integrity("commit_merged_record", Some(new_id));
1129 Ok(new_id)
1130 }
1131
1132 fn overwrite_unused_records(&self) -> bool {
1133 overwrite_unused_records_locked::<T>(self)
1134 }
1135
1136 fn as_any(&self) -> &dyn Any {
1137 self
1138 }
1139}
1140
1141pub(crate) struct MutableStateInner<T: Clone + 'static> {
1142 pub(crate) state: Arc<SnapshotMutableState<T>>,
1143 pub(crate) watchers: RefCell<HashMap<ScopeId, RcWeak<RecomposeScopeInner>>>,
1144 runtime: RuntimeHandle,
1145 state_id: Cell<Option<StateId>>,
1146}
1147
1148fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<RcWeak<dyn Fn()>>>) {
1149 let callbacks_snapshot = std::mem::take(&mut *callbacks.borrow_mut());
1150 let mut live = Vec::with_capacity(callbacks_snapshot.len());
1151 for callback in callbacks_snapshot {
1152 let Some(callback) = callback.upgrade() else {
1153 continue;
1154 };
1155 callback();
1156 live.push(callback);
1157 }
1158 let mut registered = callbacks.borrow_mut();
1159 registered.retain(|callback| callback.upgrade().is_some());
1160 for callback in live {
1161 let callback = Rc::downgrade(&callback);
1162 if !registered
1163 .iter()
1164 .any(|registered| registered.ptr_eq(&callback))
1165 {
1166 registered.push(callback);
1167 }
1168 }
1169}
1170
1171fn register_subscriber_callback(
1172 callbacks: &RefCell<Vec<RcWeak<dyn Fn()>>>,
1173 callback: &Rc<dyn Fn()>,
1174) {
1175 let callback_weak = Rc::downgrade(callback);
1176 let mut callbacks = callbacks.borrow_mut();
1177 callbacks.retain(|callback| callback.upgrade().is_some());
1178 if !callbacks
1179 .iter()
1180 .any(|registered| registered.ptr_eq(&callback_weak))
1181 {
1182 callbacks.push(callback_weak);
1183 }
1184}
1185
1186fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, RcWeak<RecomposeScopeInner>>) {
1187 let len = watchers.len();
1188 let capacity = watchers.capacity();
1189 if capacity > len.saturating_mul(4).max(32) {
1190 watchers.shrink_to_fit();
1191 }
1192}
1193
1194impl<T: Clone + 'static> MutableStateInner<T> {
1195 pub(crate) fn new_with_policy(
1196 value: T,
1197 runtime: RuntimeHandle,
1198 policy: Arc<dyn MutationPolicy<T>>,
1199 ) -> Self {
1200 Self {
1201 state: SnapshotMutableState::new_in_arc(value, policy),
1202 watchers: RefCell::new(HashMap::default()),
1203 runtime,
1204 state_id: Cell::new(None),
1205 }
1206 }
1207
1208 pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1209 self.state_id.set(Some(state_id));
1210 let runtime_handle = self.runtime.clone();
1211 self.state.add_apply_observer(Box::new(move || {
1212 let runtime = runtime_handle.clone();
1213 runtime_handle.enqueue_ui_task(Box::new(move || {
1214 runtime.with_state_arena(|arena| {
1215 let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1216 inner.invalidate_watchers();
1217 });
1218 });
1219 }));
1220 }));
1221 }
1222
1223 fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1224 let mut watchers = self.watchers.borrow_mut();
1225 let before = watchers.len();
1226 watchers.retain(|_, existing| existing.upgrade().is_some());
1227 self.state.remove_scope_observers(before - watchers.len());
1228 let registered = match watchers.get(&scope.id()) {
1229 Some(_) => false,
1230 _ => {
1231 watchers.insert(scope.id(), scope.downgrade());
1232 true
1233 }
1234 };
1235 drop(watchers);
1236 let became_subscribed = registered && self.state.add_scope_observer();
1237 (registered, became_subscribed)
1238 }
1239
1240 fn has_subscribers(&self) -> bool {
1241 let mut watchers = self.watchers.borrow_mut();
1242 let before = watchers.len();
1243 watchers.retain(|_, existing| existing.upgrade().is_some());
1244 self.state.remove_scope_observers(before - watchers.len());
1245 self.state.has_subscribers()
1246 }
1247
1248 pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1249 let mut watchers = self.watchers.borrow_mut();
1250 let removed = if watchers
1251 .get(&scope_id)
1252 .is_some_and(|weak| weak.upgrade().is_none())
1253 {
1254 watchers.remove(&scope_id);
1255 shrink_watchers_if_sparse(&mut watchers);
1256 true
1257 } else {
1258 false
1259 };
1260 drop(watchers);
1261 self.state.remove_scope_observers(usize::from(removed));
1262 }
1263
1264 fn state_id(&self) -> Option<StateId> {
1265 self.state_id.get()
1266 }
1267
1268 fn invalidate_watchers(&self) {
1269 let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1270 let mut watchers = self.watchers.borrow_mut();
1271 let before = watchers.len();
1272 let mut live = Vec::with_capacity(watchers.len());
1273 watchers.retain(|_, scope| {
1274 if let Some(inner) = scope.upgrade() {
1275 live.push(RecomposeScope { inner });
1276 true
1277 } else {
1278 false
1279 }
1280 });
1281 let removed_count = before - watchers.len();
1282 shrink_watchers_if_sparse(&mut watchers);
1283 (live, removed_count)
1284 };
1285 self.state.remove_scope_observers(removed_count);
1286
1287 for watcher in watchers {
1288 debug_record_scope_invalidation::<T>(watcher.id(), self.state_id.get());
1289 if let Some(state_id) = self.state_id.get() {
1290 watcher.invalidate_from_state(state_id);
1291 } else {
1292 watcher.invalidate();
1293 }
1294 }
1295 }
1296}
1297
1298impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1299 fn drop(&mut self) {
1300 self.state
1301 .remove_scope_observers(self.watchers.get_mut().len());
1302 }
1303}
1304
1305fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1306 let Some(Some(scope)) =
1307 with_current_composer_opt(|composer| composer.current_state_invalidation_scope())
1308 else {
1309 return;
1310 };
1311 let (registered, became_subscribed) = inner.register_scope(&scope);
1312 if registered {
1313 if let Some(state_id) = inner.state_id() {
1314 scope.record_state_subscription(state_id);
1315 }
1316 if became_subscribed {
1317 inner.state.notify_subscribers();
1318 }
1319 }
1320}
1321
1322trait StateArenaHandle<T: Clone + 'static> {
1323 fn state_id(&self) -> StateId;
1324 fn runtime_id(&self) -> runtime::RuntimeId;
1325
1326 fn runtime_handle(&self) -> RuntimeHandle {
1327 runtime::runtime_handle_by_id(self.runtime_id())
1328 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1329 }
1330
1331 fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1332 runtime::runtime_handle_by_id(self.runtime_id())
1333 }
1334
1335 fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1336 self.runtime_handle()
1337 .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1338 }
1339
1340 fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1341 self.runtime_handle_opt()?
1342 .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1343 }
1344}
1345
1346pub struct State<T: Clone + 'static> {
1348 id: StateId,
1349 runtime_id: runtime::RuntimeId,
1350 _marker: PhantomData<fn() -> T>,
1351}
1352
1353pub struct MutableState<T: Clone + 'static> {
1359 id: StateId,
1360 runtime_id: runtime::RuntimeId,
1361 _marker: PhantomData<fn() -> T>,
1362}
1363
1364#[derive(Clone)]
1366pub struct OwnedMutableState<T: Clone + 'static> {
1367 state: MutableState<T>,
1368 _lease: Rc<runtime::StateHandleLease>,
1369 _marker: PhantomData<fn() -> T>,
1370}
1371
1372impl<T: Clone + 'static> PartialEq for State<T> {
1373 fn eq(&self, other: &Self) -> bool {
1374 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1375 }
1376}
1377
1378impl<T: Clone + 'static> Eq for State<T> {}
1379
1380impl<T: Clone + 'static> PartialEq for MutableState<T> {
1381 fn eq(&self, other: &Self) -> bool {
1382 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1383 }
1384}
1385
1386impl<T: Clone + 'static> Eq for MutableState<T> {}
1387
1388impl<T: Clone + 'static> Copy for State<T> {}
1389
1390impl<T: Clone + 'static> Clone for State<T> {
1391 fn clone(&self) -> Self {
1392 *self
1393 }
1394}
1395
1396impl<T: Clone + 'static> Copy for MutableState<T> {}
1397
1398impl<T: Clone + 'static> Clone for MutableState<T> {
1399 fn clone(&self) -> Self {
1400 *self
1401 }
1402}
1403
1404impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1405 fn state_id(&self) -> StateId {
1406 self.id
1407 }
1408
1409 fn runtime_id(&self) -> runtime::RuntimeId {
1410 self.runtime_id
1411 }
1412}
1413
1414impl<T: Clone + 'static> State<T> {
1415 fn subscribe_current_scope(&self) {
1416 self.with_inner(register_current_state_scope::<T>);
1417 }
1418
1419 pub fn is_alive(&self) -> bool {
1420 self.try_with_inner(|_| ()).is_some()
1421 }
1422
1423 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1424 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1425 }
1426
1427 pub fn try_value(&self) -> Option<T> {
1428 self.try_with_inner(|inner| inner.state.try_get())?
1429 }
1430
1431 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1432 let value = self.with_inner(|inner| inner.state.get());
1433 self.subscribe_current_scope();
1434 f(&value)
1435 }
1436
1437 pub fn value(&self) -> T {
1438 let value = self.with_inner(|inner| inner.state.get());
1439 self.subscribe_current_scope();
1440 value
1441 }
1442
1443 pub fn get(&self) -> T {
1444 self.value()
1445 }
1446
1447 pub fn has_subscribers(&self) -> bool {
1448 self.with_inner(MutableStateInner::has_subscribers)
1449 }
1450
1451 pub fn on_subscriber(&self, callback: Rc<dyn Fn()>) {
1452 self.with_inner(|inner| {
1453 inner
1454 .state
1455 .subscriber_callback(callback, inner.has_subscribers())
1456 });
1457 }
1458
1459 pub fn subscription_hold(&self) -> StateSubscriptionHold {
1471 StateSubscriptionHold {
1472 _lease: self
1473 .try_with_inner(|inner| inner.state.observation_lease())
1474 .flatten(),
1475 }
1476 }
1477}
1478
1479pub struct StateSubscriptionHold {
1482 _lease: Option<Box<dyn Any>>,
1483}
1484
1485impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1486 fn state_id(&self) -> StateId {
1487 self.id
1488 }
1489
1490 fn runtime_id(&self) -> runtime::RuntimeId {
1491 self.runtime_id
1492 }
1493}
1494
1495impl<T: Clone + 'static> MutableState<T> {
1496 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1497 runtime.alloc_persistent_state(value)
1498 }
1499
1500 fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1501 Self {
1502 id,
1503 runtime_id,
1504 _marker: PhantomData,
1505 }
1506 }
1507
1508 pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1509 Self::from_parts(lease.id(), lease.runtime().id())
1510 }
1511
1512 pub fn is_alive(&self) -> bool {
1513 self.try_with_inner(|_| ()).is_some()
1514 }
1515
1516 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1517 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1518 }
1519
1520 pub fn try_value(&self) -> Option<T> {
1521 self.try_with_inner(|inner| inner.state.try_get())?
1522 }
1523
1524 pub fn as_state(&self) -> State<T> {
1525 State {
1526 id: self.id,
1527 runtime_id: self.runtime_id,
1528 _marker: PhantomData,
1529 }
1530 }
1531
1532 pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1533 let lease = self
1534 .runtime_handle_opt()?
1535 .retain_state_lease(self.state_id())?;
1536 Some(OwnedMutableState {
1537 state: *self,
1538 _lease: lease,
1539 _marker: PhantomData,
1540 })
1541 }
1542
1543 pub fn retain(&self) -> OwnedMutableState<T> {
1544 self.try_retain()
1545 .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1546 }
1547
1548 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1549 let value = self.with_inner(|inner| inner.state.get());
1550 self.subscribe_current_scope();
1551 f(&value)
1552 }
1553
1554 pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1555 let runtime = self.runtime_handle();
1556 runtime.assert_ui_thread();
1557 runtime.with_state_arena(|arena| {
1558 arena.with_typed::<T, R>(self.state_id(), |inner| {
1559 let mut value = inner.state.get();
1560 let tracker = UpdateScope::new(inner.state.id());
1561 let result = f(&mut value);
1562 let wrote_elsewhere = tracker.finish();
1563 if !wrote_elsewhere && inner.state.set(value) {
1564 inner.invalidate_watchers();
1565 }
1566 result
1567 })
1568 })
1569 }
1570
1571 pub fn replace(&self, value: T) {
1572 let Some(runtime) = self.runtime_handle_opt() else {
1573 log::debug!(
1574 "MutableState::replace skipped: runtime {:?} dropped",
1575 self.runtime_id()
1576 );
1577 return;
1578 };
1579 runtime.assert_ui_thread();
1580 let replaced = runtime
1581 .try_with_state_arena(|arena| {
1582 arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1583 if inner.state.set(value) {
1584 inner.invalidate_watchers();
1585 }
1586 })
1587 })
1588 .flatten();
1589 if replaced.is_none() {
1590 log::debug!(
1591 "MutableState::replace skipped: state cell released (slot={}, gen={})",
1592 self.state_id().slot(),
1593 self.state_id().generation(),
1594 );
1595 }
1596 }
1597
1598 pub fn set_value(&self, value: T) {
1599 self.replace(value);
1600 }
1601
1602 pub fn set(&self, value: T) {
1603 self.replace(value);
1604 }
1605
1606 pub fn value(&self) -> T {
1607 let value = self.with_inner(|inner| inner.state.get());
1608 self.subscribe_current_scope();
1609 value
1610 }
1611
1612 pub fn get(&self) -> T {
1613 self.value()
1614 }
1615
1616 pub fn get_non_reactive(&self) -> T {
1617 self.with_inner(|inner| inner.state.get())
1618 }
1619
1620 #[doc(hidden)]
1621 pub fn runtime_state_id(&self) -> StateId {
1622 self.state_id()
1623 }
1624
1625 #[doc(hidden)]
1626 pub fn subscribe_current_scope_only(&self) {
1627 self.subscribe_current_scope();
1628 }
1629
1630 fn subscribe_current_scope(&self) {
1631 self.with_inner(register_current_state_scope::<T>);
1632 }
1633
1634 #[cfg(test)]
1635 pub(crate) fn watcher_count(&self) -> usize {
1636 self.with_inner(|inner| inner.watchers.borrow().len())
1637 }
1638
1639 #[cfg(test)]
1640 pub(crate) fn watcher_capacity(&self) -> usize {
1641 self.with_inner(|inner| inner.watchers.borrow().capacity())
1642 }
1643
1644 #[cfg(test)]
1645 pub(crate) fn subscriber_callback_count(&self) -> usize {
1646 self.with_inner(|inner| inner.state.subscriber_callback_count())
1647 }
1648
1649 #[cfg(test)]
1650 pub(crate) fn state_id_for_test(&self) -> StateId {
1651 self.state_id()
1652 }
1653
1654 #[cfg(test)]
1655 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1656 self.as_state().subscribe_scope_for_test(scope);
1657 }
1658}
1659
1660impl<T: Clone + 'static> OwnedMutableState<T> {
1661 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1662 let lease = runtime.alloc_state(value);
1663 Self {
1664 state: MutableState::from_lease(&lease),
1665 _lease: lease,
1666 _marker: PhantomData,
1667 }
1668 }
1669
1670 pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1671 where
1672 T: PartialEq,
1673 {
1674 Self::with_runtime_and_policy(value, runtime, Arc::new(StructuralEqual))
1675 }
1676
1677 pub(crate) fn with_runtime_and_policy(
1678 value: T,
1679 runtime: RuntimeHandle,
1680 policy: Arc<dyn MutationPolicy<T>>,
1681 ) -> Self {
1682 let lease = runtime.alloc_state_with_policy(value, policy);
1683 Self {
1684 state: MutableState::from_lease(&lease),
1685 _lease: lease,
1686 _marker: PhantomData,
1687 }
1688 }
1689
1690 pub fn handle(&self) -> MutableState<T> {
1691 self.state
1692 }
1693
1694 pub fn as_state(&self) -> State<T> {
1695 self.state.as_state()
1696 }
1697}
1698
1699impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1700 type Target = MutableState<T>;
1701
1702 fn deref(&self) -> &Self::Target {
1703 &self.state
1704 }
1705}
1706
1707#[cfg(test)]
1708impl<T: Clone + 'static> State<T> {
1709 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1710 self.with_inner(|inner| {
1711 let (registered, became_subscribed) = inner.register_scope(scope);
1712 if registered {
1713 if let Some(state_id) = inner.state_id() {
1714 scope.record_state_subscription(state_id);
1715 }
1716 if became_subscribed {
1717 inner.state.notify_subscribers();
1718 }
1719 }
1720 });
1721 }
1722}
1723
1724impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1725 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1726 if let Some(value) = self.try_value() {
1727 f.debug_struct("MutableState")
1728 .field("value", &value)
1729 .finish()
1730 } else {
1731 f.write_str("MutableState { value: <unavailable> }")
1732 }
1733 }
1734}
1735
1736#[derive(Clone)]
1737pub struct SnapshotStateList<T: Clone + 'static> {
1738 state: OwnedMutableState<Vec<T>>,
1739}
1740
1741impl<T: Clone + 'static> SnapshotStateList<T> {
1742 pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1743 where
1744 I: IntoIterator<Item = T>,
1745 {
1746 let initial: Vec<T> = values.into_iter().collect();
1747 Self {
1748 state: OwnedMutableState::with_runtime(initial, runtime),
1749 }
1750 }
1751
1752 pub fn as_state(&self) -> State<Vec<T>> {
1753 self.state.as_state()
1754 }
1755
1756 pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1757 self.state.handle()
1758 }
1759
1760 pub fn len(&self) -> usize {
1761 self.state.with(|values| values.len())
1762 }
1763
1764 pub fn is_empty(&self) -> bool {
1765 self.len() == 0
1766 }
1767
1768 pub fn to_vec(&self) -> Vec<T> {
1769 self.state.with(|values| values.clone())
1770 }
1771
1772 pub fn iter(&self) -> Vec<T> {
1773 self.to_vec()
1774 }
1775
1776 pub fn get(&self, index: usize) -> T {
1777 self.state.with(|values| values[index].clone())
1778 }
1779
1780 pub fn get_opt(&self, index: usize) -> Option<T> {
1781 self.state.with(|values| values.get(index).cloned())
1782 }
1783
1784 pub fn first(&self) -> Option<T> {
1785 self.get_opt(0)
1786 }
1787
1788 pub fn last(&self) -> Option<T> {
1789 self.state.with(|values| values.last().cloned())
1790 }
1791
1792 pub fn push(&self, value: T) {
1793 self.state.update(|values| values.push(value));
1794 }
1795
1796 pub fn extend<I>(&self, iter: I)
1797 where
1798 I: IntoIterator<Item = T>,
1799 {
1800 self.state.update(|values| values.extend(iter));
1801 }
1802
1803 pub fn insert(&self, index: usize, value: T) {
1804 self.state.update(|values| values.insert(index, value));
1805 }
1806
1807 pub fn set(&self, index: usize, value: T) -> T {
1808 self.state
1809 .update(|values| std::mem::replace(&mut values[index], value))
1810 }
1811
1812 pub fn remove(&self, index: usize) -> T {
1813 self.state.update(|values| values.remove(index))
1814 }
1815
1816 pub fn pop(&self) -> Option<T> {
1817 self.state.update(|values| values.pop())
1818 }
1819
1820 pub fn clear(&self) {
1821 self.state.replace(Vec::new());
1822 }
1823
1824 pub fn retain<F>(&self, mut predicate: F)
1825 where
1826 F: FnMut(&T) -> bool,
1827 {
1828 self.state
1829 .update(|values| values.retain(|value| predicate(value)));
1830 }
1831
1832 pub fn replace_with<I>(&self, iter: I)
1833 where
1834 I: IntoIterator<Item = T>,
1835 {
1836 self.state.replace(iter.into_iter().collect());
1837 }
1838}
1839
1840impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
1841 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1842 let contents = self.to_vec();
1843 f.debug_struct("SnapshotStateList")
1844 .field("values", &contents)
1845 .finish()
1846 }
1847}
1848
1849#[derive(Clone)]
1850pub struct SnapshotStateMap<K, V>
1851where
1852 K: Clone + Eq + Hash + 'static,
1853 V: Clone + 'static,
1854{
1855 state: OwnedMutableState<HashMap<K, V>>,
1856}
1857
1858impl<K, V> SnapshotStateMap<K, V>
1859where
1860 K: Clone + Eq + Hash + 'static,
1861 V: Clone + 'static,
1862{
1863 pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
1864 where
1865 I: IntoIterator<Item = (K, V)>,
1866 {
1867 let map: HashMap<K, V> = pairs.into_iter().collect();
1868 Self {
1869 state: OwnedMutableState::with_runtime(map, runtime),
1870 }
1871 }
1872
1873 pub fn as_state(&self) -> State<HashMap<K, V>> {
1874 self.state.as_state()
1875 }
1876
1877 pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
1878 self.state.handle()
1879 }
1880
1881 pub fn len(&self) -> usize {
1882 self.state.with(|map| map.len())
1883 }
1884
1885 pub fn is_empty(&self) -> bool {
1886 self.state.with(|map| map.is_empty())
1887 }
1888
1889 pub fn contains_key(&self, key: &K) -> bool {
1890 self.state.with(|map| map.contains_key(key))
1891 }
1892
1893 pub fn get(&self, key: &K) -> Option<V> {
1894 self.state.with(|map| map.get(key).cloned())
1895 }
1896
1897 pub fn to_hash_map(&self) -> HashMap<K, V> {
1898 self.state.with(|map| map.clone())
1899 }
1900
1901 pub fn insert(&self, key: K, value: V) -> Option<V> {
1902 self.state.update(|map| map.insert(key, value))
1903 }
1904
1905 pub fn extend<I>(&self, iter: I)
1906 where
1907 I: IntoIterator<Item = (K, V)>,
1908 {
1909 self.state.update(|map| map.extend(iter));
1910 }
1911
1912 pub fn remove(&self, key: &K) -> Option<V> {
1913 self.state.update(|map| map.remove(key))
1914 }
1915
1916 pub fn clear(&self) {
1917 self.state.replace(HashMap::default());
1918 }
1919
1920 pub fn retain<F>(&self, mut predicate: F)
1921 where
1922 F: FnMut(&K, &mut V) -> bool,
1923 {
1924 self.state.update(|map| map.retain(|k, v| predicate(k, v)));
1925 }
1926}
1927
1928impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
1929where
1930 K: Clone + Eq + Hash + fmt::Debug + 'static,
1931 V: Clone + fmt::Debug + 'static,
1932{
1933 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1934 let contents = self.to_hash_map();
1935 f.debug_struct("SnapshotStateMap")
1936 .field("entries", &contents)
1937 .finish()
1938 }
1939}
1940
1941pub(crate) struct DerivedState<T: Clone + 'static> {
1942 compute: Rc<dyn Fn() -> T>,
1943 pub(crate) state: OwnedMutableState<T>,
1944}
1945
1946impl<T: Clone + 'static> DerivedState<T> {
1947 pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
1948 let initial = compute();
1949 Self {
1950 compute,
1951 state: OwnedMutableState::with_runtime(initial, runtime),
1952 }
1953 }
1954
1955 pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
1956 self.compute = compute;
1957 }
1958
1959 pub(crate) fn recompute(&self) {
1960 let value = (self.compute)();
1961 self.state.set_value(value);
1962 }
1963}
1964
1965impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
1966 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1967 if let Some(value) = self.try_value() {
1968 f.debug_struct("State").field("value", &value).finish()
1969 } else {
1970 f.write_str("State { value: <unavailable> }")
1971 }
1972 }
1973}
1974
1975#[cfg(test)]
1976mod tests {
1977 use super::*;
1978
1979 fn create_record_chain(ids: &[SnapshotId]) -> Rc<StateRecord> {
1980 let mut head: Option<Rc<StateRecord>> = None;
1981
1982 for &id in ids.iter().rev() {
1983 head = Some(StateRecord::new(id, 0i32, head));
1984 }
1985
1986 head.expect("create_record_chain called with empty ids")
1987 }
1988
1989 struct ManualState {
1990 head: Rc<StateRecord>,
1991 }
1992
1993 impl ManualState {
1994 fn new(head: Rc<StateRecord>) -> Self {
1995 Self { head }
1996 }
1997 }
1998
1999 impl StateObject for ManualState {
2000 fn object_id(&self) -> ObjectId {
2001 ObjectId(999)
2002 }
2003
2004 fn first_record(&self) -> Rc<StateRecord> {
2005 Rc::clone(&self.head)
2006 }
2007
2008 fn try_readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Option<Rc<StateRecord>> {
2009 Some(Rc::clone(&self.head))
2010 }
2011
2012 fn readable_record(&self, _: SnapshotId, _: &SnapshotIdSet) -> Rc<StateRecord> {
2013 Rc::clone(&self.head)
2014 }
2015
2016 fn prepend_state_record(&self, _: Rc<StateRecord>) {}
2017
2018 fn promote_record(&self, _: SnapshotId) -> Result<(), &'static str> {
2019 Ok(())
2020 }
2021
2022 fn as_any(&self) -> &dyn Any {
2023 self
2024 }
2025 }
2026
2027 fn poison_mutex<T>(mutex: &Mutex<T>) {
2028 let poison_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2029 let _guard = mutex
2030 .lock()
2031 .unwrap_or_else(|poisoned| poisoned.into_inner());
2032 panic!("poison snapshot state mutex for recovery test");
2033 }));
2034
2035 assert!(poison_result.is_err());
2036 }
2037
2038 #[test]
2039 fn observation_leases_drive_subscriber_liveness() {
2040 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2041 let notifications = Rc::new(Cell::new(0));
2042 let notifications_for_callback = Rc::clone(¬ifications);
2043 let callback: Rc<dyn Fn()> = Rc::new(move || {
2044 notifications_for_callback.set(notifications_for_callback.get() + 1);
2045 });
2046 state.subscriber_callback(callback.clone(), false);
2047
2048 let first = StateObject::observation_lease(&*state).expect("first observation lease");
2049 let second = StateObject::observation_lease(&*state).expect("second observation lease");
2050 assert!(state.has_subscribers());
2051 assert_eq!(notifications.get(), 1);
2052
2053 drop(first);
2054 assert!(state.has_subscribers());
2055 drop(second);
2056 assert!(!state.has_subscribers());
2057
2058 let third = StateObject::observation_lease(&*state).expect("third observation lease");
2059 assert_eq!(notifications.get(), 2);
2060 drop(third);
2061 }
2062
2063 #[test]
2064 fn snapshot_mutable_state_recovers_poisoned_weak_self_lock() {
2065 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2066
2067 poison_mutex(&state.weak_self);
2068
2069 assert_eq!(state.get(), 100);
2070 assert!(state.set(101));
2071 assert_eq!(state.get(), 101);
2072 }
2073
2074 #[test]
2075 fn snapshot_mutable_state_recovers_poisoned_apply_observer_lock() {
2076 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2077 let calls = Rc::new(Cell::new(0usize));
2078 let observed_calls = Rc::clone(&calls);
2079
2080 poison_mutex(&state.apply_observers);
2081
2082 state.add_apply_observer(Box::new(move || {
2083 observed_calls.set(observed_calls.get() + 1);
2084 }));
2085 state.notify_applied();
2086
2087 assert_eq!(calls.get(), 1);
2088 }
2089
2090 #[test]
2091 fn snapshot_mutable_state_promote_missing_record_returns_error() {
2092 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2093 let missing_snapshot = usize::MAX - 17;
2094
2095 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2096 StateObject::promote_record(&*state, missing_snapshot)
2097 }));
2098
2099 assert!(
2100 matches!(result, Ok(Err("missing child record"))),
2101 "missing child record should be reported through Result, got {result:?}"
2102 );
2103 }
2104
2105 #[test]
2106 fn snapshot_mutable_state_promote_wrong_record_type_returns_error() {
2107 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2108 let child_snapshot = usize::MAX - 31;
2109 let wrong_record = StateRecord::new(child_snapshot, "wrong type", None);
2110 StateObject::prepend_state_record(&*state, wrong_record);
2111
2112 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2113 StateObject::promote_record(&*state, child_snapshot)
2114 }));
2115
2116 assert!(
2117 matches!(result, Ok(Err("child record value missing or wrong type"))),
2118 "wrong child record type should be reported through Result, got {result:?}"
2119 );
2120 }
2121
2122 #[test]
2123 fn snapshot_mutable_state_commit_wrong_record_type_returns_error() {
2124 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2125 let merged = StateRecord::new(usize::MAX - 43, "wrong type", None);
2126
2127 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2128 StateObject::commit_merged_record(&*state, merged)
2129 }));
2130
2131 assert!(
2132 matches!(result, Ok(Err("merged record value missing or wrong type"))),
2133 "wrong merged record type should be reported through Result, got {result:?}"
2134 );
2135 }
2136
2137 #[test]
2138 fn snapshot_mutable_state_merge_wrong_record_type_returns_none() {
2139 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2140 let previous = StateRecord::new(usize::MAX - 51, 1i32, None);
2141 let current = StateRecord::new(usize::MAX - 52, "wrong type", None);
2142 let applied = StateRecord::new(usize::MAX - 53, 2i32, None);
2143
2144 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
2145 StateObject::merge_records(&*state, previous, current, applied)
2146 }));
2147
2148 match result {
2149 Ok(None) => {}
2150 Ok(Some(_)) => panic!("wrong merge record type unexpectedly produced a merged record"),
2151 Err(_) => panic!("wrong merge record type should not panic"),
2152 }
2153 }
2154
2155 #[test]
2156 fn test_used_locked_finds_invalid_snapshot() {
2157 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2158 let invalid_rec = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, Some(tail));
2159 let head = StateRecord::new(10, 0i32, Some(invalid_rec.clone()));
2160
2161 let result = used_locked(&head);
2162 assert!(result.is_some());
2163 assert_eq!(result.unwrap().snapshot_id(), INVALID_SNAPSHOT_ID);
2164 }
2165
2166 #[test]
2167 fn test_used_locked_finds_obscured_record() {
2168 crate::snapshot_pinning::reset_pinning_table();
2169
2170 let pin_handle = crate::snapshot_pinning::track_pinning(10, &SnapshotIdSet::EMPTY);
2171
2172 let oldest = StateRecord::new(2, 0i32, None);
2173 let newer = StateRecord::new(5, 0i32, Some(oldest.clone()));
2174 let head = StateRecord::new(100, 0i32, Some(newer));
2175
2176 let result = used_locked(&head);
2177
2178 assert!(result.is_some());
2179 let reused = result.unwrap();
2180 assert_eq!(
2181 reused.snapshot_id(),
2182 2,
2183 "Should return the oldest obscured record"
2184 );
2185
2186 crate::snapshot_pinning::release_pinning(pin_handle);
2187 }
2188
2189 #[test]
2190 fn test_used_locked_no_reusable_record() {
2191 crate::snapshot_pinning::reset_pinning_table();
2192
2193 let high_id = allocate_record_id() + 1000;
2194 let head = create_record_chain(&[high_id, high_id + 1, high_id + 2]);
2195
2196 let result = used_locked(&head);
2197 assert!(
2198 result.is_none(),
2199 "Should find no reusable records when all are recent"
2200 );
2201 }
2202
2203 #[test]
2204 fn test_used_locked_single_old_record() {
2205 crate::snapshot_pinning::reset_pinning_table();
2206
2207 let old = StateRecord::new(2, 0i32, None);
2208 let head = StateRecord::new(100, 0i32, Some(old));
2209
2210 let result = used_locked(&head);
2211 assert!(result.is_none(), "Single old record should not be reused");
2212 }
2213
2214 #[test]
2215 fn test_readable_record_for_preexisting() {
2216 let head = create_record_chain(&[PREEXISTING_SNAPSHOT_ID]);
2217 let invalid = SnapshotIdSet::EMPTY;
2218
2219 let result = readable_record_for(&head, 10, &invalid);
2220 assert!(result.is_some());
2221 assert_eq!(result.unwrap().snapshot_id(), PREEXISTING_SNAPSHOT_ID);
2222 }
2223
2224 #[test]
2225 fn test_readable_record_for_picks_highest_valid() {
2226 let head = create_record_chain(&[10, 5, PREEXISTING_SNAPSHOT_ID]);
2227 let invalid = SnapshotIdSet::EMPTY;
2228
2229 let result = readable_record_for(&head, 10, &invalid);
2230 assert!(result.is_some());
2231 assert_eq!(result.unwrap().snapshot_id(), 10);
2232
2233 let result = readable_record_for(&head, 7, &invalid);
2234 assert!(result.is_some());
2235 assert_eq!(result.unwrap().snapshot_id(), 5);
2236 }
2237
2238 #[test]
2239 fn test_new_overwritable_record_locked_reuses_invalid() {
2240 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2241
2242 let current_head = state.first_record();
2243 let invalid_rec = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, current_head.next());
2244 current_head.set_next(Some(invalid_rec.clone()));
2245
2246 let result = new_overwritable_record_locked(&*state);
2247
2248 assert!(Rc::ptr_eq(&result, &invalid_rec));
2249 assert_eq!(result.snapshot_id(), SNAPSHOT_ID_MAX);
2250 }
2251
2252 #[test]
2253 fn test_new_overwritable_record_locked_creates_new() {
2254 crate::snapshot_pinning::reset_pinning_table();
2255
2256 let _pin_handle = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2257
2258 let state = SnapshotMutableState::new_in_arc(100i32, Arc::new(NeverEqual));
2259 let old_head = state.first_record();
2260
2261 let result = new_overwritable_record_locked(&*state);
2262
2263 assert_eq!(result.snapshot_id(), SNAPSHOT_ID_MAX);
2264
2265 let new_head = state.first_record();
2266 assert!(
2267 Rc::ptr_eq(&new_head, &result),
2268 "new_head ({:p}) should equal result ({:p})",
2269 Rc::as_ptr(&new_head),
2270 Rc::as_ptr(&result)
2271 );
2272
2273 assert!(result.next().is_some());
2274 assert!(Rc::ptr_eq(&result.next().unwrap(), &old_head));
2275 }
2276
2277 #[test]
2278 fn test_writable_record_reuses_invalid_record() {
2279 crate::snapshot_pinning::reset_pinning_table();
2280
2281 let state = SnapshotMutableState::new_in_arc(7i32, Arc::new(NeverEqual));
2282
2283 let head = state.first_record();
2284 let invalid = StateRecord::new(INVALID_SNAPSHOT_ID, 0i32, head.next());
2285 head.set_next(Some(invalid.clone()));
2286
2287 let snapshot_id = allocate_record_id();
2288 let result = state.writable_record(snapshot_id, &SnapshotIdSet::EMPTY);
2289
2290 assert!(
2291 Rc::ptr_eq(&result, &invalid),
2292 "Expected writable_record to reuse the INVALID record"
2293 );
2294 assert_eq!(result.snapshot_id(), snapshot_id);
2295 result.with_value(|value: &i32| {
2296 assert_eq!(*value, 7, "Reused record should copy the readable value");
2297 });
2298 assert!(!result.is_tombstone());
2299 }
2300
2301 #[test]
2302 fn test_writable_record_creates_new_when_reuse_disallowed() {
2303 crate::snapshot_pinning::reset_pinning_table();
2304 let pin = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2305
2306 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2307 let original_head = state.first_record();
2308 let preexisting = original_head
2309 .next()
2310 .expect("preexisting record should exist for newly created state");
2311
2312 let snapshot_id = allocate_record_id();
2313 let result = state.writable_record(snapshot_id, &SnapshotIdSet::EMPTY);
2314
2315 assert!(
2316 !Rc::ptr_eq(&result, &original_head),
2317 "Should not reuse the current head when reuse is disallowed"
2318 );
2319 assert!(
2320 !Rc::ptr_eq(&result, &preexisting),
2321 "Should not reuse the PREEXISTING record"
2322 );
2323 assert_eq!(result.snapshot_id(), snapshot_id);
2324 result.with_value(|value: &i32| assert_eq!(*value, 42));
2325
2326 let new_head = state.first_record();
2327 assert!(
2328 Rc::ptr_eq(&new_head, &result),
2329 "Newly created record should become the head of the chain"
2330 );
2331
2332 crate::snapshot_pinning::release_pinning(pin);
2333 }
2334
2335 #[test]
2336 fn test_state_record_clear_for_reuse() {
2337 let record = StateRecord::new(10, 42i32, None);
2338
2339 record.with_value(|val: &i32| {
2340 assert_eq!(*val, 42);
2341 });
2342
2343 record.clear_for_reuse();
2344
2345 assert_eq!(record.snapshot_id(), 10);
2346 }
2347
2348 #[test]
2349 fn test_overwrite_unused_records_no_old_records() {
2350 crate::snapshot_pinning::reset_pinning_table();
2351
2352 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2353
2354 let _pin = crate::snapshot_pinning::track_pinning(1, &SnapshotIdSet::EMPTY);
2355
2356 let should_retain = state.overwrite_unused_records();
2357
2358 assert!(
2359 should_retain,
2360 "Should retain multiple records when none are old enough"
2361 );
2362
2363 let mut cursor = Some(state.first_record());
2364 while let Some(record) = cursor {
2365 assert_ne!(record.snapshot_id(), INVALID_SNAPSHOT_ID);
2366 cursor = record.next();
2367 }
2368 }
2369
2370 #[test]
2371 fn test_overwrite_unused_records_basic_cleanup() {
2372 crate::snapshot_pinning::reset_pinning_table();
2373
2374 let rec1 = StateRecord::new(100, 1i32, None);
2375 let rec2 = StateRecord::new(200, 2i32, Some(rec1.clone()));
2376 let rec3 = StateRecord::new(300, 3i32, Some(rec2.clone()));
2377
2378 let test_state = ManualState::new(rec3.clone());
2379
2380 let _pin = crate::snapshot_pinning::track_pinning(1000, &SnapshotIdSet::EMPTY);
2381
2382 let result = overwrite_unused_records_locked::<i32>(&test_state);
2383
2384 assert_eq!(rec3.snapshot_id(), 300);
2385 assert_eq!(rec2.snapshot_id(), INVALID_SNAPSHOT_ID);
2386 assert_eq!(rec1.snapshot_id(), INVALID_SNAPSHOT_ID);
2387
2388 assert!(!result);
2389 }
2390
2391 #[test]
2392 fn test_overwrite_unused_records_single_record_only() {
2393 crate::snapshot_pinning::reset_pinning_table();
2394
2395 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2396
2397 let head = state.first_record();
2398 head.set_next(None);
2399
2400 let should_retain = state.overwrite_unused_records();
2401
2402 assert!(!should_retain, "Single record should return false");
2403 }
2404
2405 #[test]
2406 fn snapshot_state_try_get_reports_missing_visible_record_without_panicking() {
2407 crate::snapshot_pinning::reset_pinning_table();
2408
2409 let state = SnapshotMutableState::new_in_arc(42i32, Arc::new(NeverEqual));
2410 let head = state.first_record();
2411 head.set_snapshot_id(SNAPSHOT_ID_MAX);
2412 head.set_next(None);
2413
2414 assert_eq!(state.try_get(), None);
2415 }
2416
2417 #[test]
2418 fn test_overwrite_unused_records_clears_values() {
2419 crate::snapshot_pinning::reset_pinning_table();
2420
2421 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2422 let old_rec1 = StateRecord::new(2, 999i32, Some(tail.clone()));
2423 let old_rec2 = StateRecord::new(3, 888i32, Some(old_rec1.clone()));
2424 let head = StateRecord::new(150, 42i32, Some(old_rec2.clone()));
2425 let state = ManualState::new(head.clone());
2426
2427 old_rec1.with_value(|val: &i32| {
2428 assert_eq!(*val, 999);
2429 });
2430
2431 let _pin = crate::snapshot_pinning::track_pinning(100, &SnapshotIdSet::EMPTY);
2432 overwrite_unused_records_locked::<i32>(&state);
2433
2434 assert_eq!(old_rec1.snapshot_id(), INVALID_SNAPSHOT_ID);
2435 }
2436
2437 #[test]
2438 fn test_overwrite_unused_records_mixed_old_and_new() {
2439 crate::snapshot_pinning::reset_pinning_table();
2440
2441 let preexisting = StateRecord::new(PREEXISTING_SNAPSHOT_ID, 0i32, None);
2442 let rec2 = StateRecord::new(2, 100i32, Some(preexisting.clone()));
2443 let rec5 = StateRecord::new(5, 100i32, Some(rec2.clone()));
2444 let rec50 = StateRecord::new(50, 100i32, Some(rec5.clone()));
2445 let head = StateRecord::new(120, 100i32, Some(rec50.clone()));
2446 let state = ManualState::new(head.clone());
2447
2448 let _pin = crate::snapshot_pinning::track_pinning(40, &SnapshotIdSet::EMPTY);
2449
2450 let should_retain = overwrite_unused_records_locked::<i32>(&state);
2451 assert!(should_retain);
2452
2453 assert_eq!(rec50.snapshot_id(), 50);
2454 assert_eq!(rec5.snapshot_id(), 5);
2455 assert_eq!(rec2.snapshot_id(), INVALID_SNAPSHOT_ID);
2456 }
2457
2458 #[test]
2459 fn test_readable_record_for_skips_invalid_set() {
2460 let head = create_record_chain(&[10, 5, PREEXISTING_SNAPSHOT_ID]);
2461 let invalid = SnapshotIdSet::new().set(5);
2462
2463 let result = readable_record_for(&head, 10, &invalid);
2464 assert!(result.is_some());
2465 assert_eq!(result.unwrap().snapshot_id(), 10);
2466
2467 let result = readable_record_for(&head, 7, &invalid);
2468 assert!(result.is_some());
2469 assert_eq!(result.unwrap().snapshot_id(), PREEXISTING_SNAPSHOT_ID);
2470 }
2471
2472 #[test]
2473 fn test_assign_value_copies_int() {
2474 let source = StateRecord::new(10, 42i32, None);
2475 let target = StateRecord::new(20, 0i32, None);
2476
2477 target.assign_value::<i32>(&source).expect("copy int value");
2478
2479 target.with_value(|val: &i32| {
2480 assert_eq!(*val, 42);
2481 });
2482
2483 source.with_value(|val: &i32| {
2484 assert_eq!(*val, 42);
2485 });
2486
2487 assert_eq!(source.snapshot_id(), 10);
2488 assert_eq!(target.snapshot_id(), 20);
2489 }
2490
2491 #[test]
2492 fn test_assign_value_copies_string() {
2493 let source = StateRecord::new(10, "hello".to_string(), None);
2494 let target = StateRecord::new(20, "world".to_string(), None);
2495
2496 target
2497 .assign_value::<String>(&source)
2498 .expect("copy string value");
2499
2500 target.with_value(|val: &String| {
2501 assert_eq!(val, "hello");
2502 });
2503
2504 source.with_value(|val: &String| {
2505 assert_eq!(val, "hello");
2506 });
2507 }
2508
2509 #[test]
2510 fn test_assign_value_reports_cleared_source() {
2511 let source = StateRecord::new(10, 42i32, None);
2512 let target = StateRecord::new(20, 0i32, None);
2513
2514 source.clear_value();
2515
2516 assert_eq!(
2517 target.assign_value::<i32>(&source),
2518 Err(StateRecordValueError::MissingOrWrongType {
2519 expected: std::any::type_name::<i32>(),
2520 })
2521 );
2522 assert_eq!(target.with_value(|val: &i32| *val), 0);
2523 }
2524
2525 #[test]
2526 fn test_assign_value_overwrites_existing_value() {
2527 let source = StateRecord::new(10, 100i32, None);
2528 let target = StateRecord::new(20, 999i32, None);
2529
2530 target.with_value(|val: &i32| {
2531 assert_eq!(*val, 999);
2532 });
2533
2534 target
2535 .assign_value::<i32>(&source)
2536 .expect("overwrite int value");
2537
2538 target.with_value(|val: &i32| {
2539 assert_eq!(*val, 100);
2540 });
2541 }
2542
2543 #[test]
2544 fn test_assign_value_with_custom_type() {
2545 #[derive(Clone, PartialEq, Debug)]
2546 struct Point {
2547 x: f64,
2548 y: f64,
2549 }
2550
2551 let source = StateRecord::new(10, Point { x: 1.5, y: 2.5 }, None);
2552 let target = StateRecord::new(20, Point { x: 0.0, y: 0.0 }, None);
2553
2554 target
2555 .assign_value::<Point>(&source)
2556 .expect("copy point value");
2557
2558 target.with_value(|val: &Point| {
2559 assert_eq!(val, &Point { x: 1.5, y: 2.5 });
2560 });
2561 }
2562
2563 #[test]
2564 fn test_assign_value_self_assignment() {
2565 let record = StateRecord::new(10, 42i32, None);
2566
2567 record
2568 .assign_value::<i32>(&record)
2569 .expect("self-assign int value");
2570
2571 record.with_value(|val: &i32| {
2572 assert_eq!(*val, 42);
2573 });
2574 }
2575
2576 #[test]
2577 fn event_loop_writes_keep_the_record_chain_bounded() {
2578 crate::snapshot_pinning::reset_pinning_table();
2579 let state = SnapshotMutableState::new_in_arc(0.0f32, Arc::new(NeverEqual));
2580
2581 let mut lens = Vec::new();
2582 for event in 0..3000usize {
2583 crate::run_in_mutable_snapshot(|| {
2584 state.set(event as f32);
2585 })
2586 .expect("event snapshot applies");
2587 let _ = state.get();
2588 if event % 500 == 499 {
2589 lens.push(state.record_chain_debug().len());
2590 }
2591 }
2592
2593 let final_len = *lens.last().expect("sampled chain lengths");
2594 assert!(
2595 final_len <= 16,
2596 "record chain grew without bound across event-loop writes: {lens:?}"
2597 );
2598 }
2599
2600 #[test]
2601 fn test_assign_value_with_vec() {
2602 let source = StateRecord::new(10, vec![1, 2, 3, 4, 5], None);
2603 let target = StateRecord::new(20, Vec::<i32>::new(), None);
2604
2605 target
2606 .assign_value::<Vec<i32>>(&source)
2607 .expect("copy vec value");
2608
2609 target.with_value(|val: &Vec<i32>| {
2610 assert_eq!(val, &vec![1, 2, 3, 4, 5]);
2611 });
2612
2613 source.replace_value(vec![10, 20]);
2614 target.with_value(|val: &Vec<i32>| {
2615 assert_eq!(val, &vec![1, 2, 3, 4, 5]);
2616 });
2617 }
2618}