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