1use std::{
2 any::Any,
3 cell::{Cell, RefCell},
4 fmt,
5 hash::Hash,
6 marker::PhantomData,
7 ops::Deref,
8 rc::{Rc, Weak as RcWeak},
9 sync::{Arc, Mutex, MutexGuard, PoisonError, Weak},
10};
11
12use crate::{
13 RecomposeScope, RecomposeScopeInner, RuntimeHandle, ScopeId, StateId,
14 collections::map::{HashMap, HashSet},
15 debug_trace::debug_record_scope_invalidation,
16 runtime,
17 snapshot_id_set::{SnapshotId, SnapshotIdSet},
18 snapshot_pinning::lowest_pinned_snapshot,
19 snapshot_v2::{
20 AnySnapshot, GlobalSnapshot, advance_global_snapshot, allocate_record_id, current_snapshot,
21 },
22 with_current_composer_opt,
23};
24
25pub(crate) const PREEXISTING_SNAPSHOT_ID: SnapshotId = 1;
26
27const INVALID_SNAPSHOT_ID: SnapshotId = 0;
28
29const SNAPSHOT_ID_MAX: SnapshotId = usize::MAX;
30
31#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug, Default)]
32pub struct ObjectId(pub(crate) usize);
33
34impl ObjectId {
35 pub(crate) fn new<T: ?Sized + 'static>(object: &Arc<T>) -> Self {
36 Self(Arc::as_ptr(object) as *const () as usize)
37 }
38
39 #[inline]
40 pub(crate) fn as_usize(self) -> usize {
41 self.0
42 }
43}
44
45pub struct StateRecord {
46 snapshot_id: Cell<SnapshotId>,
47 tombstone: Cell<bool>,
48 next: Cell<Option<Rc<StateRecord>>>,
49 value: RefCell<Option<Box<dyn Any>>>,
50}
51
52#[derive(Debug)]
53struct StateReadFailure {
54 state_id: ObjectId,
55 snapshot_id: SnapshotId,
56 fresh_snapshot_id: SnapshotId,
57 fresh_invalid: SnapshotIdSet,
58 record_chain: Vec<(SnapshotId, bool)>,
59}
60
61impl std::fmt::Display for StateReadFailure {
62 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63 write!(
64 f,
65 "Reading a state that was created after the snapshot was taken or in a snapshot that has not yet been applied\n\
66 state={:?}, snapshot_id={}, fresh_snapshot_id={}, fresh_invalid={:?}\n\
67 record_chain={:?}",
68 self.state_id,
69 self.snapshot_id,
70 self.fresh_snapshot_id,
71 self.fresh_invalid,
72 self.record_chain
73 )
74 }
75}
76
77#[derive(Debug, Clone, Copy, Eq, PartialEq)]
78pub(crate) enum StateRecordValueError {
79 MissingOrWrongType { expected: &'static str },
80}
81
82impl StateRecord {
83 pub(crate) fn new<T: Any>(
84 snapshot_id: SnapshotId,
85 value: T,
86 next: Option<Rc<StateRecord>>,
87 ) -> Rc<Self> {
88 Rc::new(Self {
89 snapshot_id: Cell::new(snapshot_id),
90 tombstone: Cell::new(false),
91 next: Cell::new(next),
92 value: RefCell::new(Some(Box::new(value))),
93 })
94 }
95
96 #[inline]
97 pub(crate) fn snapshot_id(&self) -> SnapshotId {
98 self.snapshot_id.get()
99 }
100
101 #[inline]
102 pub(crate) fn set_snapshot_id(&self, id: SnapshotId) {
103 self.snapshot_id.set(id);
104 }
105
106 #[inline]
107 pub(crate) fn next(&self) -> Option<Rc<StateRecord>> {
108 self.next.take().inspect(|record| {
109 self.next.set(Some(Rc::clone(record)));
110 })
111 }
112
113 #[inline]
114 pub(crate) fn set_next(&self, next: Option<Rc<StateRecord>>) {
115 self.next.set(next);
116 }
117
118 #[inline]
119 pub(crate) fn is_tombstone(&self) -> bool {
120 self.tombstone.get()
121 }
122
123 #[inline]
124 pub(crate) fn set_tombstone(&self, tombstone: bool) {
125 self.tombstone.set(tombstone);
126 }
127
128 pub(crate) fn clear_value(&self) {
129 self.value.borrow_mut().take();
130 }
131
132 pub(crate) fn replace_value<T: Any>(&self, new_value: T) {
133 *self.value.borrow_mut() = Some(Box::new(new_value));
134 }
135
136 pub(crate) fn with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> R {
137 self.try_with_value(f)
138 .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"))
139 }
140
141 pub(crate) fn try_with_value<T: Any, R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
142 let guard = self.value.borrow();
143 let value = guard.as_ref().and_then(|boxed| boxed.downcast_ref::<T>())?;
144 Some(f(value))
145 }
146
147 #[cfg(test)]
148 pub(crate) fn clear_for_reuse(&self) {
149 self.clear_value();
150 }
151
152 pub(crate) fn assign_value<T: Any + Clone>(
153 &self,
154 source: &StateRecord,
155 ) -> Result<(), StateRecordValueError> {
156 let cloned_value = source
157 .try_with_value(|value: &T| value.clone())
158 .ok_or_else(|| StateRecordValueError::MissingOrWrongType {
159 expected: std::any::type_name::<T>(),
160 })?;
161 self.replace_value(cloned_value);
162 Ok(())
163 }
164}
165
166impl Drop for StateRecord {
167 fn drop(&mut self) {
168 let mut next = self.next.take();
169 while let Some(node) = next {
170 match Rc::try_unwrap(node) {
171 Ok(record) => {
172 next = record.next.take();
173 }
174 Err(_) => {
175 break;
176 }
177 }
178 }
179 }
180}
181
182struct CurrentRecord {
183 head: RefCell<Rc<StateRecord>>,
184}
185
186impl CurrentRecord {
187 fn new(head: Rc<StateRecord>) -> Self {
188 Self {
189 head: RefCell::new(head),
190 }
191 }
192
193 fn clone_head(&self) -> Rc<StateRecord> {
194 self.head.borrow().clone()
195 }
196
197 fn replace(&self, new_head: Rc<StateRecord>) {
198 *self.head.borrow_mut() = new_head;
199 }
200
201 fn prepend(&self, record: Rc<StateRecord>) {
202 let current_head = self.clone_head();
203 record.set_next(Some(current_head));
204 self.replace(record);
205 }
206}
207
208#[inline]
209fn record_is_valid_for(
210 record: &Rc<StateRecord>,
211 snapshot_id: SnapshotId,
212 invalid: &SnapshotIdSet,
213) -> bool {
214 if record.is_tombstone() {
215 return false;
216 }
217
218 let candidate = record.snapshot_id();
219 if candidate == INVALID_SNAPSHOT_ID || candidate > snapshot_id {
220 return false;
221 }
222
223 candidate == snapshot_id || !invalid.get(candidate)
224}
225
226pub(crate) fn readable_record_for(
227 head: &Rc<StateRecord>,
228 snapshot_id: SnapshotId,
229 invalid: &SnapshotIdSet,
230) -> Option<Rc<StateRecord>> {
231 let mut best: Option<Rc<StateRecord>> = None;
232 let mut cursor = Some(Rc::clone(head));
233
234 while let Some(record) = cursor {
235 if record_is_valid_for(&record, snapshot_id, invalid) {
236 let replace = best
237 .as_ref()
238 .is_none_or(|current| current.snapshot_id() < record.snapshot_id());
239 if replace {
240 best = Some(Rc::clone(&record));
241 }
242 }
243 cursor = record.next();
244 }
245
246 best
247}
248
249fn find_youngest_or<F>(head: &Rc<StateRecord>, predicate: F) -> Rc<StateRecord>
250where
251 F: Fn(&Rc<StateRecord>) -> bool,
252{
253 let mut current = Some(Rc::clone(head));
254 let mut youngest = Rc::clone(head);
255
256 while let Some(record) = current {
257 if predicate(&record) {
258 return record;
259 }
260 if youngest.snapshot_id() < record.snapshot_id() {
261 youngest = Rc::clone(&record);
262 }
263 current = record.next();
264 }
265
266 youngest
267}
268
269pub(crate) fn used_locked(head: &Rc<StateRecord>) -> Option<Rc<StateRecord>> {
270 let mut current = Some(Rc::clone(head));
271 let mut valid_record: Option<Rc<StateRecord>> = None;
272
273 let reuse_limit = lowest_pinned_snapshot().map_or_else(
274 || allocate_record_id().saturating_sub(1),
275 |lowest| lowest.saturating_sub(1),
276 );
277
278 let invalid = SnapshotIdSet::EMPTY;
279
280 while let Some(record) = current {
281 let current_id = record.snapshot_id();
282
283 if current_id == PREEXISTING_SNAPSHOT_ID {
284 current = record.next();
285 continue;
286 }
287
288 if current_id == INVALID_SNAPSHOT_ID {
289 return Some(record);
290 }
291
292 if record.is_tombstone() && current_id < reuse_limit {
293 return Some(record);
294 }
295
296 if record_is_valid_for(&record, reuse_limit, &invalid) {
297 if let Some(ref existing) = valid_record {
298 return Some(if current_id < existing.snapshot_id() {
299 record
300 } else {
301 Rc::clone(existing)
302 });
303 } else {
304 valid_record = Some(record.clone());
305 }
306 }
307
308 current = record.next();
309 }
310
311 None
312}
313
314pub(crate) fn new_overwritable_record_locked(state: &dyn StateObject) -> Rc<StateRecord> {
315 let state_head = state.first_record();
316
317 if let Some(reusable) = used_locked(&state_head) {
318 reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
319 return reusable;
320 }
321
322 let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
323
324 state.prepend_state_record(Rc::clone(&new_record));
325
326 new_record
327}
328
329pub(crate) fn new_overwritable_record_as_head_locked(state: &dyn StateObject) -> Rc<StateRecord> {
330 let head = state.first_record();
331
332 if let Some(reusable) = used_locked(&head) {
333 reusable.set_snapshot_id(SNAPSHOT_ID_MAX);
334
335 if !Rc::ptr_eq(&head, &reusable) {
336 let mut cursor = Some(Rc::clone(&head));
337 let mut unlinked = false;
338
339 while let Some(node) = cursor {
340 let next = node.next();
341 if let Some(next_record) = next {
342 if Rc::ptr_eq(&next_record, &reusable) {
343 node.set_next(reusable.next());
344 unlinked = true;
345 break;
346 }
347 cursor = Some(next_record);
348 } else {
349 break;
350 }
351 }
352
353 if !unlinked {
354 debug_assert!(
355 false,
356 "new_overwritable_record_as_head_locked: reusable record not found in chain"
357 );
358 let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
359 state.prepend_state_record(Rc::clone(&new_record));
360 return new_record;
361 }
362
363 state.prepend_state_record(Rc::clone(&reusable));
364 }
365
366 return reusable;
367 }
368
369 let new_record = StateRecord::new(SNAPSHOT_ID_MAX, (), None);
370 state.prepend_state_record(Rc::clone(&new_record));
371 new_record
372}
373
374pub(crate) fn overwrite_unused_records_locked<T: Any + Clone>(state: &dyn StateObject) -> bool {
375 let head = state.first_record();
376 let mut current = Some(Rc::clone(&head));
377 let mut overwrite_record: Option<Rc<StateRecord>> = None;
378 let mut valid_record: Option<Rc<StateRecord>> = None;
379
380 let reuse_limit =
381 lowest_pinned_snapshot().unwrap_or_else(crate::snapshot_v2::peek_next_snapshot_id);
382
383 let mut retained_records = 0;
384
385 while let Some(record) = current {
386 let current_id = record.snapshot_id();
387
388 if current_id == INVALID_SNAPSHOT_ID {
389 } else if current_id < reuse_limit {
390 if valid_record.is_none() {
391 valid_record = Some(Rc::clone(&record));
392 retained_records += 1;
393 } else {
394 let Some(valid) = valid_record.as_ref() else {
395 valid_record = Some(Rc::clone(&record));
396 retained_records += 1;
397 current = record.next();
398 continue;
399 };
400 let record_to_overwrite = if current_id < valid.snapshot_id() {
401 Rc::clone(&record)
402 } else {
403 let to_overwrite = Rc::clone(valid);
404 valid_record = Some(Rc::clone(&record));
405 to_overwrite
406 };
407
408 let source_record = overwrite_record.get_or_insert_with(|| {
409 find_youngest_or(&head, |r| r.snapshot_id() >= reuse_limit)
410 });
411
412 record_to_overwrite.set_snapshot_id(INVALID_SNAPSHOT_ID);
413 if let Err(error) = record_to_overwrite.assign_value::<T>(source_record) {
414 log::error!(
415 "snapshot cleanup could not copy retained state record value for state {:?}: {:?}",
416 state.object_id(),
417 error
418 );
419 }
420 }
421 } else {
422 retained_records += 1;
423 }
424
425 current = record.next();
426 }
427
428 retained_records > 1
429}
430
431fn active_snapshot() -> AnySnapshot {
432 current_snapshot().unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()))
433}
434
435pub(crate) trait MutationPolicy<T>: Send + Sync {
436 fn equivalent(&self, a: &T, b: &T) -> bool;
437 fn merge(&self, _previous: &T, _current: &T, _applied: &T) -> Option<T> {
438 None
439 }
440}
441
442pub(crate) struct NeverEqual;
443
444impl<T> MutationPolicy<T> for NeverEqual {
445 fn equivalent(&self, _a: &T, _b: &T) -> bool {
446 false
447 }
448}
449
450pub(crate) struct StructuralEqual;
451
452impl<T: PartialEq> MutationPolicy<T> for StructuralEqual {
453 fn equivalent(&self, a: &T, b: &T) -> bool {
454 a == b
455 }
456}
457
458pub trait StateObject: Any {
459 fn object_id(&self) -> ObjectId;
460 fn first_record(&self) -> Rc<StateRecord>;
461 fn try_readable_record(
462 &self,
463 snapshot_id: SnapshotId,
464 invalid: &SnapshotIdSet,
465 ) -> Option<Rc<StateRecord>>;
466 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord>;
467
468 fn prepend_state_record(&self, record: Rc<StateRecord>);
469
470 fn merge_records(
471 &self,
472 _previous: Rc<StateRecord>,
473 _current: Rc<StateRecord>,
474 _applied: Rc<StateRecord>,
475 ) -> Option<Rc<StateRecord>> {
476 None
477 }
478
479 fn commit_merged_record(&self, _merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
480 Err("StateObject does not support merged record commits")
481 }
482 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str>;
483
484 fn overwrite_unused_records(&self) -> bool {
485 false
486 }
487
488 fn observation_lease(&self) -> Option<Rc<dyn Any>> {
489 None
490 }
491
492 fn as_any(&self) -> &dyn Any;
493}
494
495pub(crate) struct SnapshotMutableState<T> {
496 head: CurrentRecord,
497 policy: Arc<dyn MutationPolicy<T>>,
498 id: ObjectId,
499 weak_self: Mutex<Option<Weak<Self>>>,
500 apply_observers: Mutex<Vec<Box<dyn Fn() + 'static>>>,
501 read_observation_lease: Rc<()>,
502 scope_observation_count: Cell<usize>,
503 subscriber_callbacks: RefCell<Vec<Rc<dyn Fn()>>>,
504}
505
506impl<T> SnapshotMutableState<T> {
507 fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
508 if !should_check_chain_integrity() {
509 return;
510 }
511 let head = self.head.clone_head();
512 let mut cursor = Some(head);
513 let mut seen: HashSet<usize> = HashSet::default();
514 let mut ids = Vec::new();
515
516 while let Some(record) = cursor {
517 let addr = Rc::as_ptr(&record) as usize;
518 assert!(
519 seen.insert(addr),
520 "SnapshotMutableState::{} detected duplicate/cycle at record {:p} for state {:?} (snapshot_context={:?}, chain_ids={:?})",
521 caller,
522 Rc::as_ptr(&record),
523 self.id,
524 snapshot_context,
525 ids
526 );
527 ids.push(record.snapshot_id());
528 cursor = record.next();
529 }
530
531 assert!(
532 !ids.is_empty(),
533 "SnapshotMutableState::{} finished integrity scan with empty id list for state {:?} (snapshot_context={:?})",
534 caller,
535 self.id,
536 snapshot_context
537 );
538 }
539}
540
541fn should_check_chain_integrity() -> bool {
542 #[cfg(debug_assertions)]
543 {
544 true
545 }
546
547 #[cfg(not(debug_assertions))]
548 {
549 crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
550 }
551}
552
553impl<T: Clone + 'static> SnapshotMutableState<T> {
554 fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
555 let mut chain_ids = Vec::new();
556 let mut cursor = Some(self.first_record());
557 while let Some(record) = cursor {
558 chain_ids.push((record.snapshot_id(), record.is_tombstone()));
559 cursor = record.next();
560 }
561 chain_ids
562 }
563
564 fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
565 let snapshot = active_snapshot();
566 if let Some(state) = self.upgrade_self() {
567 snapshot.record_read(&*state);
568 }
569
570 let snapshot_id = snapshot.snapshot_id();
571 let invalid = snapshot.invalid();
572
573 if let Some(record) = self.readable_for(snapshot_id, &invalid) {
574 return Ok(record);
575 }
576
577 let fresh_snapshot = active_snapshot();
578 let fresh_id = fresh_snapshot.snapshot_id();
579 let fresh_invalid = fresh_snapshot.invalid();
580
581 if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
582 return Ok(record);
583 }
584
585 let global = GlobalSnapshot::get_or_create();
586 let global_id = global.snapshot_id();
587 let global_invalid = global.invalid();
588
589 if let Some(record) = self.readable_for(global_id, &global_invalid) {
590 return Ok(record);
591 }
592
593 Err(StateReadFailure {
594 state_id: self.id,
595 snapshot_id,
596 fresh_snapshot_id: fresh_id,
597 fresh_invalid,
598 record_chain: self.record_chain_debug(),
599 })
600 }
601
602 fn readable_for(
603 &self,
604 snapshot_id: SnapshotId,
605 invalid: &SnapshotIdSet,
606 ) -> Option<Rc<StateRecord>> {
607 let head = self.first_record();
608 readable_record_for(&head, snapshot_id, invalid)
609 }
610
611 fn is_equivalent_to_readable(
612 &self,
613 snapshot_id: SnapshotId,
614 invalid: &SnapshotIdSet,
615 new_value: &T,
616 ) -> bool {
617 self.readable_for(snapshot_id, invalid)
618 .is_some_and(|record| {
619 record.with_value(|current: &T| self.policy.equivalent(current, new_value))
620 })
621 }
622
623 fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
624 let Some(readable) = self.readable_for(snapshot_id, invalid) else {
625 let current_head = self.head.clone_head();
626 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid);
627 let source = refreshed.unwrap_or_else(|| current_head.clone());
628
629 let cloned_value = source.with_value(|value: &T| value.clone());
630 let new_head = StateRecord::new(snapshot_id, cloned_value, Some(current_head));
631 self.head.replace(new_head.clone());
632 self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
633 return new_head;
634 };
635
636 if readable.snapshot_id() == snapshot_id {
637 return readable;
638 }
639
640 let refreshed = {
641 let current_head = self.head.clone_head();
642 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid).unwrap_or_else(
643 || {
644 panic!(
645 "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
646 self.id, snapshot_id, invalid
647 )
648 },
649 );
650
651 if refreshed.snapshot_id() == snapshot_id {
652 return refreshed;
653 }
654
655 Rc::clone(&refreshed)
656 };
657
658 let overwritable = new_overwritable_record_locked(self);
659 if let Err(error) = overwritable.assign_value::<T>(&refreshed) {
660 log::error!(
661 "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
662 self.id,
663 error
664 );
665 }
666 overwritable.set_snapshot_id(snapshot_id);
667 overwritable.set_tombstone(false);
668
669 self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
670
671 overwritable
672 }
673
674 pub(crate) fn new_in_arc(initial: T, policy: Arc<dyn MutationPolicy<T>>) -> Arc<Self> {
675 let snapshot = active_snapshot();
676 let snapshot_id = snapshot.snapshot_id();
677
678 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
679 let head = StateRecord::new(snapshot_id, initial, Some(tail));
680
681 let mut state = Arc::new(Self {
682 head: CurrentRecord::new(head),
683 policy,
684 id: ObjectId::default(),
685 weak_self: Mutex::new(None),
686 apply_observers: Mutex::new(Vec::new()),
687 read_observation_lease: Rc::new(()),
688 scope_observation_count: Cell::new(0),
689 subscriber_callbacks: RefCell::new(Vec::new()),
690 });
691
692 let id = ObjectId::new(&state);
693 if let Some(state_inner) = Arc::get_mut(&mut state) {
694 state_inner.id = id;
695 }
696
697 *state.lock_weak_self() = Some(Arc::downgrade(&state));
698
699 state
700 }
701
702 pub(crate) fn add_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
703 self.lock_apply_observers().push(observer);
704 }
705
706 fn acquire_observation_lease(&self) -> Option<Rc<dyn Any>> {
707 let was_empty = !self.has_subscribers();
708 let lease = Rc::clone(&self.read_observation_lease);
709 if was_empty {
710 notify_subscriber_callbacks(&self.subscriber_callbacks);
711 }
712 Some(lease)
713 }
714
715 fn add_scope_observer(&self) -> bool {
716 let was_empty = !self.has_subscribers();
717 let count = self
718 .scope_observation_count
719 .get()
720 .checked_add(1)
721 .expect("state scope observation count overflow");
722 self.scope_observation_count.set(count);
723 was_empty
724 }
725
726 fn remove_scope_observers(&self, count: usize) {
727 if count == 0 {
728 return;
729 }
730 let remaining = self
731 .scope_observation_count
732 .get()
733 .checked_sub(count)
734 .expect("state scope observation count underflow");
735 self.scope_observation_count.set(remaining);
736 }
737
738 fn has_subscribers(&self) -> bool {
739 Rc::strong_count(&self.read_observation_lease) > 1 || self.scope_observation_count.get() > 0
740 }
741
742 fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
743 self.subscriber_callbacks
744 .borrow_mut()
745 .push(Rc::clone(&callback));
746 if notify {
747 callback();
748 }
749 }
750
751 fn notify_subscribers(&self) {
752 notify_subscriber_callbacks(&self.subscriber_callbacks);
753 }
754
755 fn notify_applied(&self) {
756 let observers = self.lock_apply_observers();
757 for observer in observers.iter() {
758 observer();
759 }
760 }
761
762 fn lock_weak_self(&self) -> MutexGuard<'_, Option<Weak<Self>>> {
763 self.weak_self
764 .lock()
765 .unwrap_or_else(PoisonError::into_inner)
766 }
767
768 fn lock_apply_observers(&self) -> MutexGuard<'_, Vec<Box<dyn Fn() + 'static>>> {
769 self.apply_observers
770 .lock()
771 .unwrap_or_else(PoisonError::into_inner)
772 }
773
774 fn upgrade_self(&self) -> Option<Arc<Self>> {
775 self.lock_weak_self().as_ref().and_then(Weak::upgrade)
776 }
777
778 #[inline]
779 pub(crate) fn id(&self) -> ObjectId {
780 self.id
781 }
782
783 pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
784 let record = self.readable_record_for_active_snapshot().ok()?;
785 record.try_with_value(f)
786 }
787
788 pub(crate) fn try_get(&self) -> Option<T> {
789 self.try_with_value(Clone::clone)
790 }
791
792 pub(crate) fn with_value<R>(&self, f: impl FnOnce(&T) -> R) -> R {
793 let record = self
794 .readable_record_for_active_snapshot()
795 .unwrap_or_else(|failure| panic!("{failure}"));
796 record.with_value(f)
797 }
798
799 pub(crate) fn get(&self) -> T {
800 self.with_value(Clone::clone)
801 }
802
803 pub(crate) fn set(&self, new_value: T) -> bool {
804 #[cfg(debug_assertions)]
805 {
806 let in_handler = crate::in_event_handler();
807 let in_snapshot = crate::in_applied_snapshot();
808 if in_handler && !in_snapshot {
809 log::warn!(
810 target: "cranpose::state",
811 "State modified in event handler without run_in_mutable_snapshot; \
812 this can make updates invisible to other contexts. Wrap the handler \
813 in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
814 self.id
815 );
816 }
817 }
818
819 let snapshot = active_snapshot();
820 let snapshot_id = snapshot.snapshot_id();
821
822 match &snapshot {
823 AnySnapshot::Global(global) => {
824 let invalid = snapshot.invalid();
825 if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
826 return false;
827 }
828
829 assert!(
830 !global.has_pending_children(),
831 "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
832 global.pending_children(),
833 self.id,
834 snapshot_id
835 );
836
837 let mut written_state: Option<Arc<dyn StateObject>> = None;
838 if let Some(state) = self.upgrade_self() {
839 let trait_object: Arc<dyn StateObject> = state;
840 snapshot.record_write(trait_object.clone());
841 written_state = Some(trait_object);
842 }
843 mark_update_write(self.id);
844
845 let new_id = allocate_record_id();
846 let record = new_overwritable_record_as_head_locked(self);
847 record.replace_value(new_value);
848 record.set_snapshot_id(new_id);
849 record.set_tombstone(false);
850 advance_global_snapshot(new_id);
851 self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
852
853 if !global.has_pending_children() {
854 let mut cursor = record.next();
855 while let Some(node) = cursor {
856 if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
857 node.clear_value();
858 node.set_tombstone(true);
859 }
860 cursor = node.next();
861 }
862 self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
863 }
864
865 if let Some(modified) = written_state.as_ref() {
866 crate::snapshot_v2::notify_apply_observers(
867 std::slice::from_ref(modified),
868 new_id,
869 );
870 }
871 }
872 AnySnapshot::Mutable(_)
873 | AnySnapshot::NestedMutable(_)
874 | AnySnapshot::TransparentMutable(_) => {
875 let invalid = snapshot.invalid();
876 if self.is_equivalent_to_readable(snapshot_id, &invalid, &new_value) {
877 return false;
878 }
879
880 if let Some(state) = self.upgrade_self() {
881 let trait_object: Arc<dyn StateObject> = state;
882 snapshot.record_write(trait_object);
883 }
884 mark_update_write(self.id);
885
886 let record = self.writable_record(snapshot_id, &invalid);
887 record.replace_value(new_value);
888 self.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
889 }
890 AnySnapshot::Readonly(_)
891 | AnySnapshot::NestedReadonly(_)
892 | AnySnapshot::TransparentReadonly(_) => {
893 panic!("Cannot write to a read-only snapshot");
894 }
895 }
896
897 true
898 }
899}
900
901thread_local! {
902 static ACTIVE_UPDATES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
903 static PENDING_WRITES: RefCell<HashSet<ObjectId>> = RefCell::new(HashSet::default());
904}
905
906pub(crate) struct UpdateScope {
907 id: ObjectId,
908 finished: bool,
909}
910
911impl UpdateScope {
912 pub(crate) fn new(id: ObjectId) -> Self {
913 ACTIVE_UPDATES.with(|active| {
914 active.borrow_mut().insert(id);
915 });
916 PENDING_WRITES.with(|pending| {
917 pending.borrow_mut().remove(&id);
918 });
919 Self {
920 id,
921 finished: false,
922 }
923 }
924
925 pub(crate) fn finish(mut self) -> bool {
926 self.finished = true;
927 ACTIVE_UPDATES.with(|active| {
928 active.borrow_mut().remove(&self.id);
929 });
930 PENDING_WRITES.with(|pending| pending.borrow_mut().remove(&self.id))
931 }
932}
933
934impl Drop for UpdateScope {
935 fn drop(&mut self) {
936 if self.finished {
937 return;
938 }
939 ACTIVE_UPDATES.with(|active| {
940 active.borrow_mut().remove(&self.id);
941 });
942 PENDING_WRITES.with(|pending| {
943 pending.borrow_mut().remove(&self.id);
944 });
945 }
946}
947
948fn mark_update_write(id: ObjectId) {
949 ACTIVE_UPDATES.with(|active| {
950 if active.borrow().contains(&id) {
951 PENDING_WRITES.with(|pending| {
952 pending.borrow_mut().insert(id);
953 });
954 }
955 });
956}
957
958impl<T: Clone + 'static> SnapshotMutableState<T> {
959 fn try_readable_record(
960 &self,
961 snapshot_id: SnapshotId,
962 invalid: &SnapshotIdSet,
963 ) -> Option<Rc<StateRecord>> {
964 self.readable_for(snapshot_id, invalid)
965 }
966}
967
968impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
969 fn object_id(&self) -> ObjectId {
970 self.id
971 }
972
973 fn first_record(&self) -> Rc<StateRecord> {
974 self.head.clone_head()
975 }
976
977 fn try_readable_record(
978 &self,
979 snapshot_id: SnapshotId,
980 invalid: &SnapshotIdSet,
981 ) -> Option<Rc<StateRecord>> {
982 self.try_readable_record(snapshot_id, invalid)
983 }
984
985 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
986 self.try_readable_record(snapshot_id, invalid)
987 .unwrap_or_else(|| {
988 panic!(
989 "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
990 self.id, snapshot_id
991 )
992 })
993 }
994
995 fn prepend_state_record(&self, record: Rc<StateRecord>) {
996 self.head.prepend(record);
997 }
998
999 fn observation_lease(&self) -> Option<Rc<dyn Any>> {
1000 self.acquire_observation_lease()
1001 }
1002
1003 fn merge_records(
1004 &self,
1005 previous: Rc<StateRecord>,
1006 current: Rc<StateRecord>,
1007 applied: Rc<StateRecord>,
1008 ) -> Option<Rc<StateRecord>> {
1009 let Some(current_value) = current.try_with_value(|value: &T| value.clone()) else {
1010 log::error!(
1011 "SnapshotMutableState::merge_records current record value missing or wrong type (state {:?}, current_id={})",
1012 self.id,
1013 current.snapshot_id()
1014 );
1015 return None;
1016 };
1017 let Some(applied_value) = applied.try_with_value(|value: &T| value.clone()) else {
1018 log::error!(
1019 "SnapshotMutableState::merge_records applied record value missing or wrong type (state {:?}, applied_id={})",
1020 self.id,
1021 applied.snapshot_id()
1022 );
1023 return None;
1024 };
1025 if self.policy.equivalent(¤t_value, &applied_value) {
1026 return Some(current);
1027 }
1028
1029 let Some(previous_value) = previous.try_with_value(|value: &T| value.clone()) else {
1030 log::error!(
1031 "SnapshotMutableState::merge_records previous record value missing or wrong type (state {:?}, previous_id={})",
1032 self.id,
1033 previous.snapshot_id()
1034 );
1035 return None;
1036 };
1037 let merged = self
1038 .policy
1039 .merge(&previous_value, ¤t_value, &applied_value)?;
1040
1041 Some(StateRecord::new(applied.snapshot_id(), merged, None))
1042 }
1043
1044 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1045 let head = self.first_record();
1046 let mut cursor = Some(head);
1047 while let Some(record) = cursor {
1048 if record.snapshot_id() == child_id {
1049 let Some(cloned) = record.try_with_value(|value: &T| value.clone()) else {
1050 log::error!(
1051 "SnapshotMutableState::promote_record child record value missing or wrong type (state {:?}, child_id={})",
1052 self.id,
1053 child_id
1054 );
1055 return Err("child record value missing or wrong type");
1056 };
1057 let new_id = allocate_record_id();
1058 let promoted = new_overwritable_record_as_head_locked(self);
1059 promoted.replace_value(cloned);
1060 promoted.set_tombstone(false);
1061 promoted.set_snapshot_id(new_id);
1062 advance_global_snapshot(new_id);
1063 self.notify_applied();
1064 self.assert_chain_integrity("promote_record", Some(child_id));
1065 return Ok(());
1066 }
1067 cursor = record.next();
1068 }
1069 log::error!(
1070 "SnapshotMutableState::promote_record missing child record (state {:?}, child_id={})",
1071 self.id,
1072 child_id
1073 );
1074 Err("missing child record")
1075 }
1076
1077 fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1078 let Some(value) = merged.try_with_value(|value: &T| value.clone()) else {
1079 log::error!(
1080 "SnapshotMutableState::commit_merged_record merged record value missing or wrong type (state {:?}, merged_id={})",
1081 self.id,
1082 merged.snapshot_id()
1083 );
1084 return Err("merged record value missing or wrong type");
1085 };
1086 let new_id = allocate_record_id();
1087 let committed = new_overwritable_record_as_head_locked(self);
1088 committed.replace_value(value);
1089 committed.set_tombstone(false);
1090 committed.set_snapshot_id(new_id);
1091 advance_global_snapshot(new_id);
1092 self.notify_applied();
1093 self.assert_chain_integrity("commit_merged_record", Some(new_id));
1094 Ok(new_id)
1095 }
1096
1097 fn overwrite_unused_records(&self) -> bool {
1098 overwrite_unused_records_locked::<T>(self)
1099 }
1100
1101 fn as_any(&self) -> &dyn Any {
1102 self
1103 }
1104}
1105
1106pub(crate) struct MutableStateInner<T: Clone + 'static> {
1107 pub(crate) state: Arc<SnapshotMutableState<T>>,
1108 pub(crate) watchers: RefCell<HashMap<ScopeId, RcWeak<RecomposeScopeInner>>>,
1109 runtime: RuntimeHandle,
1110 state_id: Cell<Option<StateId>>,
1111}
1112
1113fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<Rc<dyn Fn()>>>) {
1114 let registered = callbacks.borrow().len();
1115 for index in 0..registered {
1116 let callback = callbacks.borrow().get(index).map(Rc::clone);
1117 if let Some(callback) = callback {
1118 callback();
1119 }
1120 }
1121}
1122
1123fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, RcWeak<RecomposeScopeInner>>) {
1124 let len = watchers.len();
1125 let capacity = watchers.capacity();
1126 if capacity > len.saturating_mul(4).max(32) {
1127 watchers.shrink_to_fit();
1128 }
1129}
1130
1131impl<T: Clone + 'static> MutableStateInner<T> {
1132 pub(crate) fn new_with_policy(
1133 value: T,
1134 runtime: RuntimeHandle,
1135 policy: Arc<dyn MutationPolicy<T>>,
1136 ) -> Self {
1137 Self {
1138 state: SnapshotMutableState::new_in_arc(value, policy),
1139 watchers: RefCell::new(HashMap::default()),
1140 runtime,
1141 state_id: Cell::new(None),
1142 }
1143 }
1144
1145 pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1146 self.state_id.set(Some(state_id));
1147 let runtime_handle = self.runtime.clone();
1148 self.state.add_apply_observer(Box::new(move || {
1149 let runtime = runtime_handle.clone();
1150 runtime_handle.enqueue_ui_task(Box::new(move || {
1151 runtime.with_state_arena(|arena| {
1152 let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1153 inner.invalidate_watchers();
1154 });
1155 });
1156 }));
1157 }));
1158 }
1159
1160 fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1161 let mut watchers = self.watchers.borrow_mut();
1162 let before = watchers.len();
1163 watchers.retain(|_, existing| existing.upgrade().is_some());
1164 self.state.remove_scope_observers(before - watchers.len());
1165 let registered = match watchers.get(&scope.id()) {
1166 Some(_) => false,
1167 _ => {
1168 watchers.insert(scope.id(), scope.downgrade());
1169 true
1170 }
1171 };
1172 drop(watchers);
1173 let became_subscribed = registered && self.state.add_scope_observer();
1174 (registered, became_subscribed)
1175 }
1176
1177 fn has_subscribers(&self) -> bool {
1178 let mut watchers = self.watchers.borrow_mut();
1179 let before = watchers.len();
1180 watchers.retain(|_, existing| existing.upgrade().is_some());
1181 self.state.remove_scope_observers(before - watchers.len());
1182 self.state.has_subscribers()
1183 }
1184
1185 pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1186 let mut watchers = self.watchers.borrow_mut();
1187 let removed = if watchers
1188 .get(&scope_id)
1189 .is_some_and(|weak| weak.upgrade().is_none())
1190 {
1191 watchers.remove(&scope_id);
1192 shrink_watchers_if_sparse(&mut watchers);
1193 true
1194 } else {
1195 false
1196 };
1197 drop(watchers);
1198 self.state.remove_scope_observers(usize::from(removed));
1199 }
1200
1201 fn state_id(&self) -> Option<StateId> {
1202 self.state_id.get()
1203 }
1204
1205 fn invalidate_watchers(&self) {
1206 let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1207 let mut watchers = self.watchers.borrow_mut();
1208 let before = watchers.len();
1209 let mut live = Vec::with_capacity(watchers.len());
1210 watchers.retain(|_, scope| {
1211 if let Some(inner) = scope.upgrade() {
1212 live.push(RecomposeScope { inner });
1213 true
1214 } else {
1215 false
1216 }
1217 });
1218 let removed_count = before - watchers.len();
1219 shrink_watchers_if_sparse(&mut watchers);
1220 (live, removed_count)
1221 };
1222 self.state.remove_scope_observers(removed_count);
1223
1224 for watcher in watchers {
1225 debug_record_scope_invalidation::<T>(watcher.id(), self.state_id.get());
1226 if let Some(state_id) = self.state_id.get() {
1227 watcher.invalidate_from_state(state_id);
1228 } else {
1229 watcher.invalidate();
1230 }
1231 }
1232 }
1233}
1234
1235impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1236 fn drop(&mut self) {
1237 self.state
1238 .remove_scope_observers(self.watchers.get_mut().len());
1239 }
1240}
1241
1242fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1243 let Some(Some(scope)) =
1244 with_current_composer_opt(super::composer::Composer::current_state_invalidation_scope)
1245 else {
1246 return;
1247 };
1248 let (registered, became_subscribed) = inner.register_scope(&scope);
1249 if registered {
1250 if let Some(state_id) = inner.state_id() {
1251 scope.record_state_subscription(state_id);
1252 }
1253 if became_subscribed {
1254 inner.state.notify_subscribers();
1255 }
1256 }
1257}
1258
1259trait StateArenaHandle<T: Clone + 'static> {
1260 fn state_id(&self) -> StateId;
1261 fn runtime_id(&self) -> runtime::RuntimeId;
1262
1263 fn runtime_handle(&self) -> RuntimeHandle {
1264 runtime::runtime_handle_by_id(self.runtime_id())
1265 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1266 }
1267
1268 fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1269 runtime::runtime_handle_by_id(self.runtime_id())
1270 }
1271
1272 fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1273 self.runtime_handle()
1274 .with_state_arena(|arena| arena.with_typed::<T, R>(self.state_id(), f))
1275 }
1276
1277 fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1278 self.runtime_handle_opt()?
1279 .try_with_state_arena(|arena| arena.with_typed_opt::<T, R>(self.state_id(), f))?
1280 }
1281}
1282
1283pub struct State<T: Clone + 'static> {
1285 id: StateId,
1286 runtime_id: runtime::RuntimeId,
1287 _marker: PhantomData<fn() -> T>,
1288}
1289
1290pub struct MutableState<T: Clone + 'static> {
1296 id: StateId,
1297 runtime_id: runtime::RuntimeId,
1298 _marker: PhantomData<fn() -> T>,
1299}
1300
1301#[derive(Clone)]
1303pub struct OwnedMutableState<T: Clone + 'static> {
1304 state: MutableState<T>,
1305 _lease: Rc<runtime::StateHandleLease>,
1306 _marker: PhantomData<fn() -> T>,
1307}
1308
1309impl<T: Clone + 'static> PartialEq for State<T> {
1310 fn eq(&self, other: &Self) -> bool {
1311 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1312 }
1313}
1314
1315impl<T: Clone + 'static> Eq for State<T> {}
1316
1317impl<T: Clone + 'static> PartialEq for MutableState<T> {
1318 fn eq(&self, other: &Self) -> bool {
1319 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1320 }
1321}
1322
1323impl<T: Clone + 'static> Eq for MutableState<T> {}
1324
1325impl<T: Clone + 'static> Copy for State<T> {}
1326
1327impl<T: Clone + 'static> Clone for State<T> {
1328 fn clone(&self) -> Self {
1329 *self
1330 }
1331}
1332
1333impl<T: Clone + 'static> Copy for MutableState<T> {}
1334
1335impl<T: Clone + 'static> Clone for MutableState<T> {
1336 fn clone(&self) -> Self {
1337 *self
1338 }
1339}
1340
1341impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1342 fn state_id(&self) -> StateId {
1343 self.id
1344 }
1345
1346 fn runtime_id(&self) -> runtime::RuntimeId {
1347 self.runtime_id
1348 }
1349}
1350
1351impl<T: Clone + 'static> State<T> {
1352 fn subscribe_current_scope(&self) {
1353 self.with_inner(register_current_state_scope::<T>);
1354 }
1355
1356 pub fn is_alive(&self) -> bool {
1357 self.try_with_inner(|_| ()).is_some()
1358 }
1359
1360 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1361 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1362 }
1363
1364 pub fn try_value(&self) -> Option<T> {
1365 self.try_with_inner(|inner| inner.state.try_get())?
1366 }
1367
1368 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1372 let value = self.with_inner(|inner| inner.state.get());
1373 self.subscribe_current_scope();
1374 f(&value)
1375 }
1376
1377 pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1382 let result = self.with_inner(|inner| inner.state.with_value(f));
1383 self.subscribe_current_scope();
1384 result
1385 }
1386
1387 pub fn value(&self) -> T {
1388 let value = self.with_inner(|inner| inner.state.get());
1389 self.subscribe_current_scope();
1390 value
1391 }
1392
1393 pub fn get(&self) -> T {
1394 self.value()
1395 }
1396
1397 pub fn has_subscribers(&self) -> bool {
1398 self.with_inner(MutableStateInner::has_subscribers)
1399 }
1400
1401 pub fn on_subscriber(&self, callback: impl Fn() + 'static) {
1407 let callback: Rc<dyn Fn()> = Rc::new(callback);
1408 self.with_inner(|inner| {
1409 inner
1410 .state
1411 .subscriber_callback(callback, inner.has_subscribers());
1412 });
1413 }
1414
1415 pub fn subscription_hold(&self) -> StateSubscriptionHold {
1427 StateSubscriptionHold {
1428 _lease: self
1429 .try_with_inner(|inner| inner.state.observation_lease())
1430 .flatten(),
1431 }
1432 }
1433}
1434
1435pub struct StateSubscriptionHold {
1438 _lease: Option<Rc<dyn Any>>,
1439}
1440
1441impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1442 fn state_id(&self) -> StateId {
1443 self.id
1444 }
1445
1446 fn runtime_id(&self) -> runtime::RuntimeId {
1447 self.runtime_id
1448 }
1449}
1450
1451impl<T: Clone + 'static> MutableState<T> {
1452 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1453 runtime.alloc_persistent_state(value)
1454 }
1455
1456 fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1457 Self {
1458 id,
1459 runtime_id,
1460 _marker: PhantomData,
1461 }
1462 }
1463
1464 pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1465 Self::from_parts(lease.id(), lease.runtime().id())
1466 }
1467
1468 pub fn is_alive(&self) -> bool {
1469 self.try_with_inner(|_| ()).is_some()
1470 }
1471
1472 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1473 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1474 }
1475
1476 pub fn try_value(&self) -> Option<T> {
1477 self.try_with_inner(|inner| inner.state.try_get())?
1478 }
1479
1480 pub fn as_state(&self) -> State<T> {
1481 State {
1482 id: self.id,
1483 runtime_id: self.runtime_id,
1484 _marker: PhantomData,
1485 }
1486 }
1487
1488 pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1489 let lease = self
1490 .runtime_handle_opt()?
1491 .retain_state_lease(self.state_id())?;
1492 Some(OwnedMutableState {
1493 state: *self,
1494 _lease: lease,
1495 _marker: PhantomData,
1496 })
1497 }
1498
1499 pub fn retain(&self) -> OwnedMutableState<T> {
1500 self.try_retain()
1501 .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1502 }
1503
1504 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1508 let value = self.with_inner(|inner| inner.state.get());
1509 self.subscribe_current_scope();
1510 f(&value)
1511 }
1512
1513 pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1518 let result = self.with_inner(|inner| inner.state.with_value(f));
1519 self.subscribe_current_scope();
1520 result
1521 }
1522
1523 pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1524 let runtime = self.runtime_handle();
1525 runtime.assert_ui_thread();
1526 runtime.with_state_arena(|arena| {
1527 arena.with_typed::<T, R>(self.state_id(), |inner| {
1528 let mut value = inner.state.get();
1529 let tracker = UpdateScope::new(inner.state.id());
1530 let result = f(&mut value);
1531 let wrote_elsewhere = tracker.finish();
1532 if !wrote_elsewhere && inner.state.set(value) {
1533 inner.invalidate_watchers();
1534 }
1535 result
1536 })
1537 })
1538 }
1539
1540 pub fn replace(&self, value: T) {
1541 let Some(runtime) = self.runtime_handle_opt() else {
1542 log::debug!(
1543 "MutableState::replace skipped: runtime {:?} dropped",
1544 self.runtime_id()
1545 );
1546 return;
1547 };
1548 runtime.assert_ui_thread();
1549 let replaced = runtime
1550 .try_with_state_arena(|arena| {
1551 arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1552 if inner.state.set(value) {
1553 inner.invalidate_watchers();
1554 }
1555 })
1556 })
1557 .flatten();
1558 if replaced.is_none() {
1559 log::debug!(
1560 "MutableState::replace skipped: state cell released (slot={}, gen={})",
1561 self.state_id().slot(),
1562 self.state_id().generation(),
1563 );
1564 }
1565 }
1566
1567 pub fn set_value(&self, value: T) {
1568 self.replace(value);
1569 }
1570
1571 pub fn set(&self, value: T) {
1572 self.replace(value);
1573 }
1574
1575 pub fn value(&self) -> T {
1576 let value = self.with_inner(|inner| inner.state.get());
1577 self.subscribe_current_scope();
1578 value
1579 }
1580
1581 pub fn get(&self) -> T {
1582 self.value()
1583 }
1584
1585 pub fn get_non_reactive(&self) -> T {
1586 self.with_inner(|inner| inner.state.get())
1587 }
1588
1589 #[doc(hidden)]
1590 pub fn runtime_state_id(&self) -> StateId {
1591 self.state_id()
1592 }
1593
1594 #[doc(hidden)]
1595 pub fn subscribe_current_scope_only(&self) {
1596 self.subscribe_current_scope();
1597 }
1598
1599 fn subscribe_current_scope(&self) {
1600 self.with_inner(register_current_state_scope::<T>);
1601 }
1602
1603 #[cfg(test)]
1604 pub(crate) fn watcher_count(&self) -> usize {
1605 self.with_inner(|inner| inner.watchers.borrow().len())
1606 }
1607
1608 #[cfg(test)]
1609 pub(crate) fn watcher_capacity(&self) -> usize {
1610 self.with_inner(|inner| inner.watchers.borrow().capacity())
1611 }
1612
1613 #[cfg(test)]
1614 pub(crate) fn state_id_for_test(&self) -> StateId {
1615 self.state_id()
1616 }
1617
1618 #[cfg(test)]
1619 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1620 self.as_state().subscribe_scope_for_test(scope);
1621 }
1622}
1623
1624impl<T: Clone + 'static> OwnedMutableState<T> {
1625 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1626 let lease = runtime.alloc_state(value);
1627 Self {
1628 state: MutableState::from_lease(&lease),
1629 _lease: lease,
1630 _marker: PhantomData,
1631 }
1632 }
1633
1634 pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1635 where
1636 T: PartialEq,
1637 {
1638 Self::with_runtime_and_policy(value, runtime, Arc::new(StructuralEqual))
1639 }
1640
1641 pub(crate) fn with_runtime_and_policy(
1642 value: T,
1643 runtime: RuntimeHandle,
1644 policy: Arc<dyn MutationPolicy<T>>,
1645 ) -> Self {
1646 let lease = runtime.alloc_state_with_policy(value, policy);
1647 Self {
1648 state: MutableState::from_lease(&lease),
1649 _lease: lease,
1650 _marker: PhantomData,
1651 }
1652 }
1653
1654 pub fn handle(&self) -> MutableState<T> {
1655 self.state
1656 }
1657
1658 pub fn as_state(&self) -> State<T> {
1659 self.state.as_state()
1660 }
1661}
1662
1663impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1664 type Target = MutableState<T>;
1665
1666 fn deref(&self) -> &Self::Target {
1667 &self.state
1668 }
1669}
1670
1671#[cfg(test)]
1672impl<T: Clone + 'static> State<T> {
1673 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1674 self.with_inner(|inner| {
1675 let (registered, became_subscribed) = inner.register_scope(scope);
1676 if registered {
1677 if let Some(state_id) = inner.state_id() {
1678 scope.record_state_subscription(state_id);
1679 }
1680 if became_subscribed {
1681 inner.state.notify_subscribers();
1682 }
1683 }
1684 });
1685 }
1686}
1687
1688impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1689 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1690 if let Some(value) = self.try_value() {
1691 f.debug_struct("MutableState")
1692 .field("value", &value)
1693 .finish()
1694 } else {
1695 f.write_str("MutableState { value: <unavailable> }")
1696 }
1697 }
1698}
1699
1700#[derive(Clone)]
1701pub struct SnapshotStateList<T: Clone + 'static> {
1702 state: OwnedMutableState<Vec<T>>,
1703}
1704
1705impl<T: Clone + 'static> SnapshotStateList<T> {
1706 pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1707 where
1708 I: IntoIterator<Item = T>,
1709 {
1710 let initial: Vec<T> = values.into_iter().collect();
1711 Self {
1712 state: OwnedMutableState::with_runtime(initial, runtime),
1713 }
1714 }
1715
1716 pub fn as_state(&self) -> State<Vec<T>> {
1717 self.state.as_state()
1718 }
1719
1720 pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1721 self.state.handle()
1722 }
1723
1724 pub fn len(&self) -> usize {
1725 self.state.with(Vec::len)
1726 }
1727
1728 pub fn is_empty(&self) -> bool {
1729 self.len() == 0
1730 }
1731
1732 pub fn to_vec(&self) -> Vec<T> {
1733 self.state.with(Clone::clone)
1734 }
1735
1736 pub fn iter(&self) -> Vec<T> {
1737 self.to_vec()
1738 }
1739
1740 pub fn get(&self, index: usize) -> T {
1741 self.state.with(|values| values[index].clone())
1742 }
1743
1744 pub fn get_opt(&self, index: usize) -> Option<T> {
1745 self.state.with(|values| values.get(index).cloned())
1746 }
1747
1748 pub fn first(&self) -> Option<T> {
1749 self.get_opt(0)
1750 }
1751
1752 pub fn last(&self) -> Option<T> {
1753 self.state.with(|values| values.last().cloned())
1754 }
1755
1756 pub fn push(&self, value: T) {
1757 self.state.update(|values| values.push(value));
1758 }
1759
1760 pub fn extend<I>(&self, iter: I)
1761 where
1762 I: IntoIterator<Item = T>,
1763 {
1764 self.state.update(|values| values.extend(iter));
1765 }
1766
1767 pub fn insert(&self, index: usize, value: T) {
1768 self.state.update(|values| values.insert(index, value));
1769 }
1770
1771 pub fn set(&self, index: usize, value: T) -> T {
1772 self.state
1773 .update(|values| std::mem::replace(&mut values[index], value))
1774 }
1775
1776 pub fn remove(&self, index: usize) -> T {
1777 self.state.update(|values| values.remove(index))
1778 }
1779
1780 pub fn pop(&self) -> Option<T> {
1781 self.state.update(Vec::pop)
1782 }
1783
1784 pub fn clear(&self) {
1785 self.state.replace(Vec::new());
1786 }
1787
1788 pub fn retain<F>(&self, mut predicate: F)
1789 where
1790 F: FnMut(&T) -> bool,
1791 {
1792 self.state
1793 .update(|values| values.retain(|value| predicate(value)));
1794 }
1795
1796 pub fn replace_with<I>(&self, iter: I)
1797 where
1798 I: IntoIterator<Item = T>,
1799 {
1800 self.state.replace(iter.into_iter().collect());
1801 }
1802}
1803
1804impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
1805 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1806 let contents = self.to_vec();
1807 f.debug_struct("SnapshotStateList")
1808 .field("values", &contents)
1809 .finish()
1810 }
1811}
1812
1813#[derive(Clone)]
1814pub struct SnapshotStateMap<K, V>
1815where
1816 K: Clone + Eq + Hash + 'static,
1817 V: Clone + 'static,
1818{
1819 state: OwnedMutableState<HashMap<K, V>>,
1820}
1821
1822impl<K, V> SnapshotStateMap<K, V>
1823where
1824 K: Clone + Eq + Hash + 'static,
1825 V: Clone + 'static,
1826{
1827 pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
1828 where
1829 I: IntoIterator<Item = (K, V)>,
1830 {
1831 let map: HashMap<K, V> = pairs.into_iter().collect();
1832 Self {
1833 state: OwnedMutableState::with_runtime(map, runtime),
1834 }
1835 }
1836
1837 pub fn as_state(&self) -> State<HashMap<K, V>> {
1838 self.state.as_state()
1839 }
1840
1841 pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
1842 self.state.handle()
1843 }
1844
1845 pub fn len(&self) -> usize {
1846 self.state.with(std::collections::HashMap::len)
1847 }
1848
1849 pub fn is_empty(&self) -> bool {
1850 self.state.with(std::collections::HashMap::is_empty)
1851 }
1852
1853 pub fn contains_key(&self, key: &K) -> bool {
1854 self.state.with(|map| map.contains_key(key))
1855 }
1856
1857 pub fn get(&self, key: &K) -> Option<V> {
1858 self.state.with(|map| map.get(key).cloned())
1859 }
1860
1861 pub fn to_hash_map(&self) -> HashMap<K, V> {
1862 self.state.with(Clone::clone)
1863 }
1864
1865 pub fn insert(&self, key: K, value: V) -> Option<V> {
1866 self.state.update(|map| map.insert(key, value))
1867 }
1868
1869 pub fn extend<I>(&self, iter: I)
1870 where
1871 I: IntoIterator<Item = (K, V)>,
1872 {
1873 self.state.update(|map| map.extend(iter));
1874 }
1875
1876 pub fn remove(&self, key: &K) -> Option<V> {
1877 self.state.update(|map| map.remove(key))
1878 }
1879
1880 pub fn clear(&self) {
1881 self.state.replace(HashMap::default());
1882 }
1883
1884 pub fn retain<F>(&self, mut predicate: F)
1885 where
1886 F: FnMut(&K, &mut V) -> bool,
1887 {
1888 self.state.update(|map| map.retain(|k, v| predicate(k, v)));
1889 }
1890}
1891
1892impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
1893where
1894 K: Clone + Eq + Hash + fmt::Debug + 'static,
1895 V: Clone + fmt::Debug + 'static,
1896{
1897 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1898 let contents = self.to_hash_map();
1899 f.debug_struct("SnapshotStateMap")
1900 .field("entries", &contents)
1901 .finish()
1902 }
1903}
1904
1905pub(crate) struct DerivedState<T: Clone + 'static> {
1906 compute: Rc<dyn Fn() -> T>,
1907 pub(crate) state: OwnedMutableState<T>,
1908}
1909
1910impl<T: Clone + 'static> DerivedState<T> {
1911 pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
1912 let initial = compute();
1913 Self {
1914 compute,
1915 state: OwnedMutableState::with_runtime(initial, runtime),
1916 }
1917 }
1918
1919 pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
1920 self.compute = compute;
1921 }
1922
1923 pub(crate) fn recompute(&self) {
1924 let value = (self.compute)();
1925 self.state.set_value(value);
1926 }
1927}
1928
1929impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
1930 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1931 if let Some(value) = self.try_value() {
1932 f.debug_struct("State").field("value", &value).finish()
1933 } else {
1934 f.write_str("State { value: <unavailable> }")
1935 }
1936 }
1937}
1938
1939#[cfg(test)]
1940#[path = "tests/state_tests.rs"]
1941mod tests;