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
502pub(crate) fn retire_record_value(value: Box<dyn RecordValue>) {
506 let held = crate::thread_state::try_snapshots(|state| {
507 if state.retire_depth.get() == 0 {
508 return Some(value);
509 }
510 state.retired_record_values.borrow_mut().push(value);
511 None
512 });
513 drop(held);
514}
515
516pub(crate) struct RetiredValuesScope(());
519
520impl RetiredValuesScope {
521 pub(crate) fn enter() -> Self {
522 crate::thread_state::snapshots(|state| {
523 state.retire_depth.set(state.retire_depth.get() + 1);
524 });
525 Self(())
526 }
527}
528
529impl Drop for RetiredValuesScope {
530 fn drop(&mut self) {
531 let retired = crate::thread_state::snapshots(|state| {
532 let depth = state.retire_depth.get() - 1;
533 state.retire_depth.set(depth);
534 (depth == 0).then(|| state.retired_record_values.take())
535 });
536 drop(retired);
537 }
538}
539
540pub(crate) fn overwrite_unused_records_locked(state: &dyn StateObject) -> bool {
544 let mut current = Some(state.first_record());
545 let mut valid_record: Option<Rc<StateRecord>> = None;
546
547 let reuse_limit =
548 lowest_pinned_snapshot().unwrap_or_else(crate::snapshot_v2::peek_next_snapshot_id);
549
550 let mut retained_records = 0;
551
552 while let Some(record) = current {
553 let current_id = record.snapshot_id();
554
555 if current_id == INVALID_SNAPSHOT_ID {
556 } else if current_id < reuse_limit {
557 match valid_record.as_ref() {
558 None => {
559 valid_record = Some(Rc::clone(&record));
560 retained_records += 1;
561 }
562 Some(valid) => {
563 let record_to_overwrite = if current_id < valid.snapshot_id() {
564 Rc::clone(&record)
565 } else {
566 let to_overwrite = Rc::clone(valid);
567 valid_record = Some(Rc::clone(&record));
568 to_overwrite
569 };
570 record_to_overwrite.set_snapshot_id(INVALID_SNAPSHOT_ID);
571 if let Some(value) = record_to_overwrite.clear_value() {
572 retire_record_value(value);
573 }
574 }
575 }
576 } else {
577 retained_records += 1;
578 }
579
580 current = record.next();
581 }
582
583 retained_records > 1
584}
585
586fn active_snapshot() -> AnySnapshot {
587 current_snapshot().unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()))
588}
589
590pub(crate) trait MutationPolicy<T>: Send + Sync {
591 fn equivalent(&self, a: &T, b: &T) -> bool;
592 fn merge(&self, _previous: &T, _current: &T, _applied: &T) -> Option<T> {
593 None
594 }
595}
596
597pub(crate) struct NeverEqual;
598
599impl<T> MutationPolicy<T> for NeverEqual {
600 fn equivalent(&self, _a: &T, _b: &T) -> bool {
601 false
602 }
603}
604
605pub(crate) struct StructuralEqual;
606
607impl<T: PartialEq> MutationPolicy<T> for StructuralEqual {
608 fn equivalent(&self, a: &T, b: &T) -> bool {
609 a == b
610 }
611}
612
613pub trait StateObject: Any {
614 fn object_id(&self) -> ObjectId;
615 fn first_record(&self) -> Rc<StateRecord>;
616 fn try_readable_record(
617 &self,
618 snapshot_id: SnapshotId,
619 invalid: &SnapshotIdSet,
620 ) -> Option<Rc<StateRecord>>;
621 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord>;
622
623 fn prepend_state_record(&self, record: Rc<StateRecord>);
624
625 fn merge_records(
626 &self,
627 _previous: Rc<StateRecord>,
628 _current: Rc<StateRecord>,
629 _applied: Rc<StateRecord>,
630 ) -> Option<Rc<StateRecord>> {
631 None
632 }
633
634 fn commit_merged_record(&self, _merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
635 Err("StateObject does not support merged record commits")
636 }
637 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str>;
638
639 fn overwrite_unused_records(&self) -> bool {
640 false
641 }
642
643 fn observation_lease(&self) -> Option<Rc<dyn Any>> {
644 None
645 }
646
647 fn as_any(&self) -> &dyn Any;
648}
649
650pub(crate) struct SnapshotMutableState<T> {
651 head: CurrentRecord,
652 policy: Rc<dyn MutationPolicy<T>>,
653 id: ObjectId,
654 weak_self: Weak<Self>,
655 apply_observer: ApplyObserver,
656 read_observation_lease: Rc<()>,
657 scope_observation_count: Cell<usize>,
658 subscriber_callbacks: RefCell<Vec<Rc<dyn Fn()>>>,
659}
660
661#[inline(always)]
662fn assert_record_chain(
663 head: &CurrentRecord,
664 id: ObjectId,
665 caller: &str,
666 snapshot_context: Option<SnapshotId>,
667) {
668 if should_check_chain_integrity() {
669 check_chain_integrity(&head.clone_head(), id, caller, snapshot_context);
670 }
671}
672
673enum Merge<'a, T> {
674 KeepCurrent,
675 Value(Option<T>),
676 Missing(&'static str, &'a Rc<StateRecord>),
677}
678
679type ApplyObserver = OnceCell<Box<dyn Fn() + 'static>>;
680
681fn notify_applied(observer: &ApplyObserver) {
682 if let Some(observer) = observer.get() {
683 observer();
684 }
685}
686
687fn finish_apply(
688 observer: &ApplyObserver,
689 head: &CurrentRecord,
690 id: ObjectId,
691 caller: &str,
692 snapshot_id: SnapshotId,
693) {
694 notify_applied(observer);
695 assert_record_chain(head, id, caller, Some(snapshot_id));
696}
697
698struct WriteTarget<'a> {
701 head: &'a CurrentRecord,
702 id: ObjectId,
703 state: &'a dyn StateObject,
704}
705
706struct GlobalWrite {
707 record: Rc<StateRecord>,
708 new_id: SnapshotId,
709 written: Option<Rc<dyn StateObject>>,
710}
711
712impl WriteTarget<'_> {
713 fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
714 assert_record_chain(self.head, self.id, caller, snapshot_context);
715 }
716
717 fn begin_global_write(
718 &self,
719 snapshot: &AnySnapshot,
720 global: &GlobalSnapshot,
721 snapshot_id: SnapshotId,
722 written: Option<Rc<dyn StateObject>>,
723 ) -> GlobalWrite {
724 assert!(
725 !global.has_pending_children(),
726 "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
727 global.pending_children(),
728 self.id,
729 snapshot_id
730 );
731 if let Some(state) = &written {
732 snapshot.record_write(Rc::clone(state));
733 }
734 mark_update_write(self.id);
735 let new_id = allocate_record_id();
736 let record = new_overwritable_record_as_head_locked(self.state);
737 GlobalWrite {
738 record,
739 new_id,
740 written,
741 }
742 }
743
744 fn finish_global_write(
745 &self,
746 global: &GlobalSnapshot,
747 write: GlobalWrite,
748 snapshot_id: SnapshotId,
749 ) {
750 write.record.set_snapshot_id(write.new_id);
753 write.record.set_tombstone(false);
754 advance_global_snapshot(write.new_id);
755 self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
756
757 let mut cleared: SmallVec<[Box<dyn RecordValue>; 2]> = SmallVec::new();
758 if !global.has_pending_children() {
759 let mut cursor = write.record.next();
760 while let Some(node) = cursor {
761 if !node.is_tombstone() && node.snapshot_id() != PREEXISTING_SNAPSHOT_ID {
762 cleared.extend(node.clear_value());
763 node.set_tombstone(true);
764 }
765 cursor = node.next();
766 }
767 self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
768 }
769
770 if let Some(modified) = write.written.as_ref() {
771 crate::snapshot_v2::notify_apply_observers(
772 std::slice::from_ref(modified),
773 write.new_id,
774 );
775 }
776 drop(cleared);
777 }
778
779 fn begin_child_write(
780 &self,
781 snapshot: &AnySnapshot,
782 snapshot_id: SnapshotId,
783 invalid: &SnapshotIdSet,
784 written: Option<Rc<dyn StateObject>>,
785 ) -> Rc<StateRecord> {
786 if let Some(state) = written {
787 snapshot.record_write(state);
788 }
789 mark_update_write(self.id);
790 self.writable_record(snapshot_id, invalid)
791 }
792
793 fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
794 let Some(readable) = readable_record_for(&self.head.clone_head(), snapshot_id, invalid)
795 else {
796 let current_head = self.head.clone_head();
797 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid);
798 let source = refreshed.unwrap_or_else(|| current_head.clone());
799
800 let cloned_value = source
801 .cloned_value()
802 .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"));
803 let new_head = StateRecord::with_boxed(snapshot_id, cloned_value, Some(current_head));
804 self.head.replace(new_head.clone());
805 self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
806 return new_head;
807 };
808
809 if readable.snapshot_id() == snapshot_id {
810 return readable;
811 }
812
813 let refreshed = {
814 let current_head = self.head.clone_head();
815 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid).unwrap_or_else(
816 || {
817 panic!(
818 "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
819 self.id, snapshot_id, invalid
820 )
821 },
822 );
823
824 if refreshed.snapshot_id() == snapshot_id {
825 return refreshed;
826 }
827
828 Rc::clone(&refreshed)
829 };
830
831 let overwritable = new_overwritable_record_locked(self.state);
832 if let Err(error) = overwritable.assign_value(&refreshed) {
833 log::error!(
834 "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
835 self.id,
836 error
837 );
838 }
839 overwritable.set_snapshot_id(snapshot_id);
840 overwritable.set_tombstone(false);
841
842 self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
843
844 overwritable
845 }
846}
847
848#[cold]
849#[inline(never)]
850fn check_chain_integrity(
851 head: &Rc<StateRecord>,
852 id: ObjectId,
853 caller: &str,
854 snapshot_context: Option<SnapshotId>,
855) {
856 let mut cursor = Some(Rc::clone(head));
857 let mut seen: HashSet<usize> = HashSet::default();
858 let mut ids = Vec::new();
859
860 while let Some(record) = cursor {
861 let addr = Rc::as_ptr(&record) as usize;
862 assert!(
863 seen.insert(addr),
864 "SnapshotMutableState::{caller} detected duplicate/cycle at record {:p} for state {id:?} (snapshot_context={snapshot_context:?}, chain_ids={ids:?})",
865 Rc::as_ptr(&record),
866 );
867 ids.push(record.snapshot_id());
868 cursor = record.next();
869 }
870
871 assert!(
872 !ids.is_empty(),
873 "SnapshotMutableState::{caller} finished integrity scan with empty id list for state {id:?} (snapshot_context={snapshot_context:?})"
874 );
875}
876
877fn should_check_chain_integrity() -> bool {
878 #[cfg(debug_assertions)]
879 {
880 true
881 }
882
883 #[cfg(not(debug_assertions))]
884 {
885 crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
886 }
887}
888
889impl<T: Clone + 'static> SnapshotMutableState<T> {
890 fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
891 let mut chain_ids = Vec::new();
892 let mut cursor = Some(self.first_record());
893 while let Some(record) = cursor {
894 chain_ids.push((record.snapshot_id(), record.is_tombstone()));
895 cursor = record.next();
896 }
897 chain_ids
898 }
899
900 fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
901 let snapshot = active_snapshot();
902 snapshot.record_read(self);
903
904 let snapshot_id = snapshot.snapshot_id();
905 if let Some(record) =
906 snapshot.with_invalid(|invalid| self.readable_for(snapshot_id, invalid))
907 {
908 return Ok(record);
909 }
910
911 let fresh_snapshot = active_snapshot();
912 let fresh_id = fresh_snapshot.snapshot_id();
913 let fresh_invalid = fresh_snapshot.invalid();
914
915 if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
916 return Ok(record);
917 }
918
919 let global = GlobalSnapshot::get_or_create();
920 let global_id = global.snapshot_id();
921 let global_invalid = global.invalid();
922
923 if let Some(record) = self.readable_for(global_id, &global_invalid) {
924 return Ok(record);
925 }
926
927 Err(StateReadFailure {
928 state_id: self.id,
929 snapshot_id,
930 fresh_snapshot_id: fresh_id,
931 fresh_invalid,
932 record_chain: self.record_chain_debug(),
933 })
934 }
935
936 fn readable_for(
937 &self,
938 snapshot_id: SnapshotId,
939 invalid: &SnapshotIdSet,
940 ) -> Option<Rc<StateRecord>> {
941 let head = self.first_record();
942 readable_record_for(&head, snapshot_id, invalid)
943 }
944
945 fn is_equivalent_to(&self, readable: Option<Rc<StateRecord>>, new_value: &T) -> bool {
946 readable.is_some_and(|record| {
947 record.with_value(|current: &T| self.policy.equivalent(current, new_value))
948 })
949 }
950
951 pub(crate) fn new_in_rc(initial: T, policy: Rc<dyn MutationPolicy<T>>) -> Rc<Self> {
952 let snapshot = active_snapshot();
953 let snapshot_id = snapshot.snapshot_id();
954
955 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
956 let head = StateRecord::new(snapshot_id, initial, Some(tail));
957
958 Rc::new_cyclic(|weak_self| Self {
959 head: CurrentRecord::new(head),
960 policy,
961 id: ObjectId(weak_self.as_ptr() as usize),
962 weak_self: weak_self.clone(),
963 apply_observer: OnceCell::new(),
964 read_observation_lease: Rc::new(()),
965 scope_observation_count: Cell::new(0),
966 subscriber_callbacks: RefCell::new(Vec::new()),
967 })
968 }
969
970 pub(crate) fn set_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
971 assert!(
972 self.apply_observer.set(observer).is_ok(),
973 "state observer already installed"
974 );
975 }
976
977 fn acquire_observation_lease(&self) -> Option<Rc<dyn Any>> {
978 let was_empty = !self.has_subscribers();
979 let lease = Rc::clone(&self.read_observation_lease);
980 if was_empty {
981 notify_subscriber_callbacks(&self.subscriber_callbacks);
982 }
983 Some(lease)
984 }
985
986 fn add_scope_observer(&self) -> bool {
987 let was_empty = !self.has_subscribers();
988 let count = self
989 .scope_observation_count
990 .get()
991 .checked_add(1)
992 .expect("state scope observation count overflow");
993 self.scope_observation_count.set(count);
994 was_empty
995 }
996
997 fn remove_scope_observers(&self, count: usize) {
998 if count == 0 {
999 return;
1000 }
1001 let remaining = self
1002 .scope_observation_count
1003 .get()
1004 .checked_sub(count)
1005 .expect("state scope observation count underflow");
1006 self.scope_observation_count.set(remaining);
1007 }
1008
1009 fn has_subscribers(&self) -> bool {
1010 Rc::strong_count(&self.read_observation_lease) > 1 || self.scope_observation_count.get() > 0
1011 }
1012
1013 fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
1014 self.subscriber_callbacks
1015 .borrow_mut()
1016 .push(Rc::clone(&callback));
1017 if notify {
1018 callback();
1019 }
1020 }
1021
1022 fn notify_subscribers(&self) {
1023 notify_subscriber_callbacks(&self.subscriber_callbacks);
1024 }
1025
1026 fn finish_apply(&self, caller: &str, snapshot_id: SnapshotId) {
1027 finish_apply(
1028 &self.apply_observer,
1029 &self.head,
1030 self.id,
1031 caller,
1032 snapshot_id,
1033 );
1034 }
1035
1036 fn upgrade_self(&self) -> Option<Rc<Self>> {
1037 self.weak_self.upgrade()
1038 }
1039
1040 fn written_state(&self) -> Option<Rc<dyn StateObject>> {
1041 self.upgrade_self()
1042 .map(|state| -> Rc<dyn StateObject> { state })
1043 }
1044
1045 #[inline]
1046 pub(crate) fn id(&self) -> ObjectId {
1047 self.id
1048 }
1049
1050 pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1051 let record = self.readable_record_for_active_snapshot().ok()?;
1052 record.try_with_value(f)
1053 }
1054
1055 pub(crate) fn try_get(&self) -> Option<T> {
1056 self.try_with_value(Clone::clone)
1057 }
1058
1059 pub(crate) fn with_value<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1060 let record = self
1061 .readable_record_for_active_snapshot()
1062 .unwrap_or_else(|failure| panic!("{failure}"));
1063 record.with_value(f)
1064 }
1065
1066 pub(crate) fn get(&self) -> T {
1067 self.with_value(Clone::clone)
1068 }
1069
1070 pub(crate) fn set(&self, new_value: T) -> bool {
1071 #[cfg(debug_assertions)]
1072 {
1073 let in_handler = crate::in_event_handler();
1074 let in_snapshot = crate::in_applied_snapshot();
1075 if in_handler && !in_snapshot {
1076 log::warn!(
1077 target: "cranpose::state",
1078 "State modified in event handler without run_in_mutable_snapshot; \
1079 this can make updates invisible to other contexts. Wrap the handler \
1080 in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
1081 self.id
1082 );
1083 }
1084 }
1085
1086 let snapshot = active_snapshot();
1087 let snapshot_id = snapshot.snapshot_id();
1088 let target = WriteTarget {
1089 head: &self.head,
1090 id: self.id,
1091 state: self,
1092 };
1093
1094 match &snapshot {
1095 AnySnapshot::Global(global) => {
1096 let readable =
1097 snapshot.with_invalid(|invalid| self.readable_for(snapshot_id, invalid));
1098 if self.is_equivalent_to(readable, &new_value) {
1099 return false;
1100 }
1101 let write =
1102 target.begin_global_write(&snapshot, global, snapshot_id, self.written_state());
1103 let displaced = write.record.replace_value(new_value);
1104 target.finish_global_write(global, write, snapshot_id);
1105 drop(displaced);
1106 }
1107 AnySnapshot::Mutable(_)
1108 | AnySnapshot::NestedMutable(_)
1109 | AnySnapshot::TransparentMutable(_) => {
1110 let invalid = snapshot.invalid();
1111 if self.is_equivalent_to(self.readable_for(snapshot_id, &invalid), &new_value) {
1112 return false;
1113 }
1114 let record = target.begin_child_write(
1115 &snapshot,
1116 snapshot_id,
1117 &invalid,
1118 self.written_state(),
1119 );
1120 let displaced = record.replace_value(new_value);
1121 target.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
1122 drop(displaced);
1123 }
1124 AnySnapshot::Readonly(_)
1125 | AnySnapshot::NestedReadonly(_)
1126 | AnySnapshot::TransparentReadonly(_) => {
1127 panic!("Cannot write to a read-only snapshot");
1128 }
1129 }
1130
1131 true
1132 }
1133}
1134
1135pub(crate) struct UpdateScope {
1136 id: ObjectId,
1137 finished: bool,
1138}
1139
1140impl UpdateScope {
1141 pub(crate) fn new(id: ObjectId) -> Self {
1142 crate::thread_state::snapshots(|state| {
1143 state.active_updates.borrow_mut().insert(id);
1144 state.pending_writes.borrow_mut().remove(&id);
1145 });
1146 Self {
1147 id,
1148 finished: false,
1149 }
1150 }
1151
1152 pub(crate) fn finish(mut self) -> bool {
1153 self.finished = true;
1154 crate::thread_state::snapshots(|state| {
1155 state.active_updates.borrow_mut().remove(&self.id);
1156 state.pending_writes.borrow_mut().remove(&self.id)
1157 })
1158 }
1159}
1160
1161impl Drop for UpdateScope {
1162 fn drop(&mut self) {
1163 if self.finished {
1164 return;
1165 }
1166 crate::thread_state::snapshots(|state| {
1167 state.active_updates.borrow_mut().remove(&self.id);
1168 state.pending_writes.borrow_mut().remove(&self.id);
1169 });
1170 }
1171}
1172
1173fn mark_update_write(id: ObjectId) {
1174 crate::thread_state::snapshots(|state| {
1175 if state.active_updates.borrow().contains(&id) {
1176 state.pending_writes.borrow_mut().insert(id);
1177 }
1178 });
1179}
1180
1181impl<T: Clone + 'static> SnapshotMutableState<T> {
1182 fn try_readable_record(
1183 &self,
1184 snapshot_id: SnapshotId,
1185 invalid: &SnapshotIdSet,
1186 ) -> Option<Rc<StateRecord>> {
1187 self.readable_for(snapshot_id, invalid)
1188 }
1189}
1190
1191impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
1192 fn object_id(&self) -> ObjectId {
1193 self.id
1194 }
1195
1196 fn first_record(&self) -> Rc<StateRecord> {
1197 self.head.clone_head()
1198 }
1199
1200 fn try_readable_record(
1201 &self,
1202 snapshot_id: SnapshotId,
1203 invalid: &SnapshotIdSet,
1204 ) -> Option<Rc<StateRecord>> {
1205 self.try_readable_record(snapshot_id, invalid)
1206 }
1207
1208 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
1209 self.try_readable_record(snapshot_id, invalid)
1210 .unwrap_or_else(|| {
1211 panic!(
1212 "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
1213 self.id, snapshot_id
1214 )
1215 })
1216 }
1217
1218 fn prepend_state_record(&self, record: Rc<StateRecord>) {
1219 self.head.prepend(record);
1220 }
1221
1222 fn observation_lease(&self) -> Option<Rc<dyn Any>> {
1223 self.acquire_observation_lease()
1224 }
1225
1226 fn merge_records(
1227 &self,
1228 previous: Rc<StateRecord>,
1229 current: Rc<StateRecord>,
1230 applied: Rc<StateRecord>,
1231 ) -> Option<Rc<StateRecord>> {
1232 let outcome = current
1233 .try_with_value(|current_value: &T| {
1234 applied
1235 .try_with_value(|applied_value: &T| {
1236 if self.policy.equivalent(current_value, applied_value) {
1237 return Merge::KeepCurrent;
1238 }
1239 previous
1240 .try_with_value(|previous_value: &T| {
1241 Merge::Value(self.policy.merge(
1242 previous_value,
1243 current_value,
1244 applied_value,
1245 ))
1246 })
1247 .unwrap_or(Merge::Missing("previous", &previous))
1248 })
1249 .unwrap_or(Merge::Missing("applied", &applied))
1250 })
1251 .unwrap_or(Merge::Missing("current", ¤t));
1252 match outcome {
1253 Merge::KeepCurrent => Some(current),
1254 Merge::Value(merged) => {
1255 merged.map(|value| StateRecord::new(applied.snapshot_id(), value, None))
1256 }
1257 Merge::Missing(which, record) => {
1258 log_merge_value_missing(which, self.id, record);
1259 None
1260 }
1261 }
1262 }
1263
1264 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1265 promote_record_locked(self, child_id, TypeId::of::<T>())?;
1266 self.finish_apply("promote_record", child_id);
1267 Ok(())
1268 }
1269
1270 fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1271 let new_id = commit_merged_record_locked(self, &merged, TypeId::of::<T>())?;
1272 self.finish_apply("commit_merged_record", new_id);
1273 Ok(new_id)
1274 }
1275
1276 fn overwrite_unused_records(&self) -> bool {
1277 overwrite_unused_records_locked(self)
1278 }
1279
1280 fn as_any(&self) -> &dyn Any {
1281 self
1282 }
1283}
1284
1285pub(crate) struct MutableStateInner<T: Clone + 'static> {
1286 pub(crate) state: Rc<SnapshotMutableState<T>>,
1287 pub(crate) watchers: RefCell<HashMap<ScopeId, Weak<RecomposeScopeInner>>>,
1288 runtime: RuntimeHandle,
1289 state_id: Cell<Option<StateId>>,
1290}
1291
1292fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<Rc<dyn Fn()>>>) {
1293 let registered = callbacks.borrow().len();
1294 for index in 0..registered {
1295 let callback = callbacks.borrow().get(index).map(Rc::clone);
1296 if let Some(callback) = callback {
1297 callback();
1298 }
1299 }
1300}
1301
1302fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, Weak<RecomposeScopeInner>>) {
1303 let len = watchers.len();
1304 let capacity = watchers.capacity();
1305 if capacity > len.saturating_mul(4).max(32) {
1306 watchers.shrink_to_fit();
1307 }
1308}
1309
1310impl<T: Clone + 'static> MutableStateInner<T> {
1311 pub(crate) fn new_with_policy(
1312 value: T,
1313 runtime: RuntimeHandle,
1314 policy: Rc<dyn MutationPolicy<T>>,
1315 ) -> Self {
1316 Self {
1317 state: SnapshotMutableState::new_in_rc(value, policy),
1318 watchers: RefCell::new(HashMap::default()),
1319 runtime,
1320 state_id: Cell::new(None),
1321 }
1322 }
1323
1324 pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1325 self.state_id.set(Some(state_id));
1326 let runtime_handle = self.runtime.clone();
1327 self.state.set_apply_observer(Box::new(move || {
1328 let runtime = runtime_handle.clone();
1329 runtime_handle.enqueue_ui_task(Box::new(move || {
1330 runtime.with_state_arena(|arena| {
1331 let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1332 inner.invalidate_watchers();
1333 });
1334 });
1335 }));
1336 }));
1337 }
1338
1339 fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1349 let mut watchers = self.watchers.borrow_mut();
1350 let id = scope.id();
1351 match watchers.get(&id) {
1352 Some(existing) if existing.strong_count() > 0 => return (false, false),
1353 Some(_) => {
1354 watchers.remove(&id);
1355 self.state.remove_scope_observers(1);
1356 }
1357 None => {}
1358 }
1359 let any_live = watchers.values().any(|watcher| watcher.strong_count() > 0);
1360 if !any_live || watchers.len() >= watchers.capacity() {
1361 let before = watchers.len();
1362 watchers.retain(|_, watcher| watcher.strong_count() > 0);
1363 self.state.remove_scope_observers(before - watchers.len());
1364 }
1365 watchers.insert(id, scope.downgrade());
1366 drop(watchers);
1367 (true, self.state.add_scope_observer())
1368 }
1369
1370 fn has_subscribers(&self) -> bool {
1371 let mut watchers = self.watchers.borrow_mut();
1372 let before = watchers.len();
1373 watchers.retain(|_, existing| existing.upgrade().is_some());
1374 self.state.remove_scope_observers(before - watchers.len());
1375 self.state.has_subscribers()
1376 }
1377
1378 pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1379 let mut watchers = self.watchers.borrow_mut();
1380 let removed = if watchers
1381 .get(&scope_id)
1382 .is_some_and(|weak| weak.upgrade().is_none())
1383 {
1384 watchers.remove(&scope_id);
1385 shrink_watchers_if_sparse(&mut watchers);
1386 true
1387 } else {
1388 false
1389 };
1390 drop(watchers);
1391 self.state.remove_scope_observers(usize::from(removed));
1392 }
1393
1394 fn state_id(&self) -> Option<StateId> {
1395 self.state_id.get()
1396 }
1397
1398 fn invalidate_watchers(&self) {
1399 let (watchers, removed_count): (Vec<RecomposeScope>, usize) = {
1400 let mut watchers = self.watchers.borrow_mut();
1401 let before = watchers.len();
1402 let mut live = Vec::with_capacity(watchers.len());
1403 watchers.retain(|_, scope| {
1404 if let Some(inner) = scope.upgrade() {
1405 live.push(RecomposeScope { inner });
1406 true
1407 } else {
1408 false
1409 }
1410 });
1411 let removed_count = before - watchers.len();
1412 shrink_watchers_if_sparse(&mut watchers);
1413 (live, removed_count)
1414 };
1415 self.state.remove_scope_observers(removed_count);
1416
1417 for watcher in watchers {
1418 debug_record_scope_invalidation(
1419 watcher.id(),
1420 self.state_id.get(),
1421 std::any::type_name::<T>(),
1422 );
1423 if let Some(state_id) = self.state_id.get() {
1424 watcher.invalidate_from_state(state_id);
1425 } else {
1426 watcher.invalidate();
1427 }
1428 }
1429 }
1430}
1431
1432impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1433 fn drop(&mut self) {
1434 self.state
1435 .remove_scope_observers(self.watchers.get_mut().len());
1436 }
1437}
1438
1439fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1440 let Some(Some(scope)) =
1441 with_current_composer_opt(super::composer::Composer::current_state_invalidation_scope)
1442 else {
1443 return;
1444 };
1445 let (registered, became_subscribed) = inner.register_scope(&scope);
1446 if registered {
1447 if let Some(state_id) = inner.state_id() {
1448 scope.record_state_subscription(state_id);
1449 }
1450 if became_subscribed {
1451 inner.state.notify_subscribers();
1452 }
1453 }
1454}
1455
1456trait StateArenaHandle<T: Clone + 'static> {
1457 fn state_id(&self) -> StateId;
1458 fn runtime_id(&self) -> runtime::RuntimeId;
1459
1460 fn runtime_handle(&self) -> RuntimeHandle {
1461 runtime::runtime_handle_by_id(self.runtime_id())
1462 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1463 }
1464
1465 fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1466 runtime::runtime_handle_by_id(self.runtime_id())
1467 }
1468
1469 fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1470 runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1471 arena.with_typed::<T, R>(self.state_id(), f)
1472 })
1473 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1474 }
1475
1476 fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1477 runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1478 arena.with_typed_opt::<T, R>(self.state_id(), f)
1479 })?
1480 }
1481
1482 fn read_subscribed<R>(&self, read: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1483 self.with_inner(|inner| {
1484 let result = read(inner);
1485 register_current_state_scope(inner);
1486 result
1487 })
1488 }
1489}
1490
1491pub struct State<T: Clone + 'static> {
1493 id: StateId,
1494 runtime_id: runtime::RuntimeId,
1495 _marker: PhantomData<fn() -> T>,
1496}
1497
1498pub struct MutableState<T: Clone + 'static> {
1504 id: StateId,
1505 runtime_id: runtime::RuntimeId,
1506 _marker: PhantomData<fn() -> T>,
1507}
1508
1509#[derive(Clone)]
1511pub struct OwnedMutableState<T: Clone + 'static> {
1512 state: MutableState<T>,
1513 _lease: Rc<runtime::StateHandleLease>,
1514 _marker: PhantomData<fn() -> T>,
1515}
1516
1517impl<T: Clone + 'static> PartialEq for State<T> {
1518 fn eq(&self, other: &Self) -> bool {
1519 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1520 }
1521}
1522
1523impl<T: Clone + 'static> Eq for State<T> {}
1524
1525impl<T: Clone + 'static> PartialEq for MutableState<T> {
1526 fn eq(&self, other: &Self) -> bool {
1527 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1528 }
1529}
1530
1531impl<T: Clone + 'static> PartialEq for OwnedMutableState<T> {
1534 fn eq(&self, other: &Self) -> bool {
1535 self.state == other.state
1536 }
1537}
1538
1539impl<T: Clone + 'static> Eq for MutableState<T> {}
1540
1541impl<T: Clone + 'static> Copy for State<T> {}
1542
1543impl<T: Clone + 'static> Clone for State<T> {
1544 fn clone(&self) -> Self {
1545 *self
1546 }
1547}
1548
1549impl<T: Clone + 'static> Copy for MutableState<T> {}
1550
1551impl<T: Clone + 'static> Clone for MutableState<T> {
1552 fn clone(&self) -> Self {
1553 *self
1554 }
1555}
1556
1557impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1558 fn state_id(&self) -> StateId {
1559 self.id
1560 }
1561
1562 fn runtime_id(&self) -> runtime::RuntimeId {
1563 self.runtime_id
1564 }
1565}
1566
1567impl<T: Clone + 'static> State<T> {
1568 pub fn is_alive(&self) -> bool {
1569 self.try_with_inner(|_| ()).is_some()
1570 }
1571
1572 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1573 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1574 }
1575
1576 pub fn try_value(&self) -> Option<T> {
1577 self.try_with_inner(|inner| inner.state.try_get())?
1578 }
1579
1580 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1584 let value = self.read_subscribed(|inner| inner.state.get());
1585 f(&value)
1586 }
1587
1588 pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1593 self.read_subscribed(|inner| inner.state.with_value(f))
1594 }
1595
1596 pub fn value(&self) -> T {
1597 self.read_subscribed(|inner| inner.state.get())
1598 }
1599
1600 pub fn get(&self) -> T {
1601 self.value()
1602 }
1603
1604 pub fn has_subscribers(&self) -> bool {
1605 self.with_inner(MutableStateInner::has_subscribers)
1606 }
1607
1608 pub fn on_subscriber(&self, callback: impl Fn() + 'static) {
1614 let callback: Rc<dyn Fn()> = Rc::new(callback);
1615 self.with_inner(|inner| {
1616 inner
1617 .state
1618 .subscriber_callback(callback, inner.has_subscribers());
1619 });
1620 }
1621
1622 pub fn subscription_hold(&self) -> StateSubscriptionHold {
1634 StateSubscriptionHold {
1635 _lease: self
1636 .try_with_inner(|inner| inner.state.observation_lease())
1637 .flatten(),
1638 }
1639 }
1640}
1641
1642pub struct StateSubscriptionHold {
1645 _lease: Option<Rc<dyn Any>>,
1646}
1647
1648impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1649 fn state_id(&self) -> StateId {
1650 self.id
1651 }
1652
1653 fn runtime_id(&self) -> runtime::RuntimeId {
1654 self.runtime_id
1655 }
1656}
1657
1658impl<T: Clone + 'static> MutableState<T> {
1659 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1660 runtime.alloc_persistent_state(value)
1661 }
1662
1663 fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1664 Self {
1665 id,
1666 runtime_id,
1667 _marker: PhantomData,
1668 }
1669 }
1670
1671 pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1672 Self::from_parts(lease.id(), lease.runtime_id())
1673 }
1674
1675 pub fn is_alive(&self) -> bool {
1676 self.try_with_inner(|_| ()).is_some()
1677 }
1678
1679 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1680 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1681 }
1682
1683 pub fn try_value(&self) -> Option<T> {
1684 self.try_with_inner(|inner| inner.state.try_get())?
1685 }
1686
1687 pub fn as_state(&self) -> State<T> {
1688 State {
1689 id: self.id,
1690 runtime_id: self.runtime_id,
1691 _marker: PhantomData,
1692 }
1693 }
1694
1695 pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1696 let lease = self
1697 .runtime_handle_opt()?
1698 .retain_state_lease(self.state_id())?;
1699 Some(OwnedMutableState {
1700 state: *self,
1701 _lease: lease,
1702 _marker: PhantomData,
1703 })
1704 }
1705
1706 pub fn retain(&self) -> OwnedMutableState<T> {
1707 self.try_retain()
1708 .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1709 }
1710
1711 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1715 let value = self.read_subscribed(|inner| inner.state.get());
1716 f(&value)
1717 }
1718
1719 pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1724 self.read_subscribed(|inner| inner.state.with_value(f))
1725 }
1726
1727 pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1728 let runtime = self.runtime_handle();
1729 runtime.assert_ui_thread();
1730 runtime.with_state_arena(|arena| {
1731 arena.with_typed::<T, R>(self.state_id(), |inner| {
1732 let mut value = inner.state.get();
1733 let tracker = UpdateScope::new(inner.state.id());
1734 let result = f(&mut value);
1735 let wrote_elsewhere = tracker.finish();
1736 if !wrote_elsewhere && inner.state.set(value) {
1737 inner.invalidate_watchers();
1738 }
1739 result
1740 })
1741 })
1742 }
1743
1744 pub fn replace(&self, value: T) {
1745 let Some(runtime) = self.runtime_handle_opt() else {
1746 log::debug!(
1747 "MutableState::replace skipped: runtime {:?} dropped",
1748 self.runtime_id()
1749 );
1750 return;
1751 };
1752 runtime.assert_ui_thread();
1753 let replaced = runtime
1754 .try_with_state_arena(|arena| {
1755 arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1756 if inner.state.set(value) {
1757 inner.invalidate_watchers();
1758 }
1759 })
1760 })
1761 .flatten();
1762 if replaced.is_none() {
1763 log::debug!(
1764 "MutableState::replace skipped: state cell released (slot={}, gen={})",
1765 self.state_id().slot(),
1766 self.state_id().generation(),
1767 );
1768 }
1769 }
1770
1771 pub fn set_value(&self, value: T) {
1772 self.replace(value);
1773 }
1774
1775 pub fn set(&self, value: T) {
1776 self.replace(value);
1777 }
1778
1779 pub fn value(&self) -> T {
1780 self.read_subscribed(|inner| inner.state.get())
1781 }
1782
1783 pub fn get(&self) -> T {
1784 self.value()
1785 }
1786
1787 pub fn get_non_reactive(&self) -> T {
1788 self.with_inner(|inner| inner.state.get())
1789 }
1790
1791 #[doc(hidden)]
1792 pub fn runtime_state_id(&self) -> StateId {
1793 self.state_id()
1794 }
1795
1796 #[doc(hidden)]
1797 pub fn subscribe_current_scope_only(&self) {
1798 self.subscribe_current_scope();
1799 }
1800
1801 fn subscribe_current_scope(&self) {
1802 self.with_inner(register_current_state_scope::<T>);
1803 }
1804
1805 #[cfg(test)]
1806 pub(crate) fn watcher_count(&self) -> usize {
1807 self.with_inner(|inner| inner.watchers.borrow().len())
1808 }
1809
1810 #[cfg(test)]
1811 pub(crate) fn watcher_capacity(&self) -> usize {
1812 self.with_inner(|inner| inner.watchers.borrow().capacity())
1813 }
1814
1815 #[cfg(test)]
1816 pub(crate) fn state_id_for_test(&self) -> StateId {
1817 self.state_id()
1818 }
1819
1820 #[cfg(test)]
1821 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1822 self.as_state().subscribe_scope_for_test(scope);
1823 }
1824}
1825
1826impl<T: Clone + 'static> OwnedMutableState<T> {
1827 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1828 let lease = runtime.alloc_state(value);
1829 Self {
1830 state: MutableState::from_lease(&lease),
1831 _lease: lease,
1832 _marker: PhantomData,
1833 }
1834 }
1835
1836 pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1837 where
1838 T: PartialEq,
1839 {
1840 Self::with_runtime_and_policy(value, runtime, Rc::new(StructuralEqual))
1841 }
1842
1843 pub(crate) fn with_runtime_and_policy(
1844 value: T,
1845 runtime: RuntimeHandle,
1846 policy: Rc<dyn MutationPolicy<T>>,
1847 ) -> Self {
1848 let lease = runtime.alloc_state_with_policy(value, policy);
1849 Self {
1850 state: MutableState::from_lease(&lease),
1851 _lease: lease,
1852 _marker: PhantomData,
1853 }
1854 }
1855
1856 pub fn handle(&self) -> MutableState<T> {
1857 self.state
1858 }
1859
1860 pub fn as_state(&self) -> State<T> {
1861 self.state.as_state()
1862 }
1863}
1864
1865impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
1866 type Target = MutableState<T>;
1867
1868 fn deref(&self) -> &Self::Target {
1869 &self.state
1870 }
1871}
1872
1873#[cfg(test)]
1874impl<T: Clone + 'static> State<T> {
1875 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1876 self.with_inner(|inner| {
1877 let (registered, became_subscribed) = inner.register_scope(scope);
1878 if registered {
1879 if let Some(state_id) = inner.state_id() {
1880 scope.record_state_subscription(state_id);
1881 }
1882 if became_subscribed {
1883 inner.state.notify_subscribers();
1884 }
1885 }
1886 });
1887 }
1888}
1889
1890impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
1891 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1892 if let Some(value) = self.try_value() {
1893 f.debug_struct("MutableState")
1894 .field("value", &value)
1895 .finish()
1896 } else {
1897 f.write_str("MutableState { value: <unavailable> }")
1898 }
1899 }
1900}
1901
1902#[derive(Clone)]
1903pub struct SnapshotStateList<T: Clone + 'static> {
1904 state: OwnedMutableState<Vec<T>>,
1905}
1906
1907impl<T: Clone + 'static> SnapshotStateList<T> {
1908 pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
1909 where
1910 I: IntoIterator<Item = T>,
1911 {
1912 let initial: Vec<T> = values.into_iter().collect();
1913 Self {
1914 state: OwnedMutableState::with_runtime(initial, runtime),
1915 }
1916 }
1917
1918 pub fn as_state(&self) -> State<Vec<T>> {
1919 self.state.as_state()
1920 }
1921
1922 pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
1923 self.state.handle()
1924 }
1925
1926 pub fn len(&self) -> usize {
1927 self.state.with(Vec::len)
1928 }
1929
1930 pub fn is_empty(&self) -> bool {
1931 self.len() == 0
1932 }
1933
1934 pub fn to_vec(&self) -> Vec<T> {
1935 self.state.with(Clone::clone)
1936 }
1937
1938 pub fn iter(&self) -> Vec<T> {
1939 self.to_vec()
1940 }
1941
1942 pub fn get(&self, index: usize) -> T {
1943 self.state.with(|values| values[index].clone())
1944 }
1945
1946 pub fn get_opt(&self, index: usize) -> Option<T> {
1947 self.state.with(|values| values.get(index).cloned())
1948 }
1949
1950 pub fn first(&self) -> Option<T> {
1951 self.get_opt(0)
1952 }
1953
1954 pub fn last(&self) -> Option<T> {
1955 self.state.with(|values| values.last().cloned())
1956 }
1957
1958 pub fn push(&self, value: T) {
1959 self.state.update(|values| values.push(value));
1960 }
1961
1962 pub fn extend<I>(&self, iter: I)
1963 where
1964 I: IntoIterator<Item = T>,
1965 {
1966 self.state.update(|values| values.extend(iter));
1967 }
1968
1969 pub fn insert(&self, index: usize, value: T) {
1970 self.state.update(|values| values.insert(index, value));
1971 }
1972
1973 pub fn set(&self, index: usize, value: T) -> T {
1974 self.state
1975 .update(|values| std::mem::replace(&mut values[index], value))
1976 }
1977
1978 pub fn remove(&self, index: usize) -> T {
1979 self.state.update(|values| values.remove(index))
1980 }
1981
1982 pub fn pop(&self) -> Option<T> {
1983 self.state.update(Vec::pop)
1984 }
1985
1986 pub fn clear(&self) {
1987 self.state.replace(Vec::new());
1988 }
1989
1990 pub fn retain<F>(&self, mut predicate: F)
1991 where
1992 F: FnMut(&T) -> bool,
1993 {
1994 self.state
1995 .update(|values| values.retain(|value| predicate(value)));
1996 }
1997
1998 pub fn replace_with<I>(&self, iter: I)
1999 where
2000 I: IntoIterator<Item = T>,
2001 {
2002 self.state.replace(iter.into_iter().collect());
2003 }
2004}
2005
2006impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
2007 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2008 let contents = self.to_vec();
2009 f.debug_struct("SnapshotStateList")
2010 .field("values", &contents)
2011 .finish()
2012 }
2013}
2014
2015#[derive(Clone)]
2016pub struct SnapshotStateMap<K, V>
2017where
2018 K: Clone + Eq + Hash + 'static,
2019 V: Clone + 'static,
2020{
2021 state: OwnedMutableState<HashMap<K, V>>,
2022}
2023
2024impl<K, V> SnapshotStateMap<K, V>
2025where
2026 K: Clone + Eq + Hash + 'static,
2027 V: Clone + 'static,
2028{
2029 pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
2030 where
2031 I: IntoIterator<Item = (K, V)>,
2032 {
2033 let map: HashMap<K, V> = pairs.into_iter().collect();
2034 Self {
2035 state: OwnedMutableState::with_runtime(map, runtime),
2036 }
2037 }
2038
2039 pub fn as_state(&self) -> State<HashMap<K, V>> {
2040 self.state.as_state()
2041 }
2042
2043 pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
2044 self.state.handle()
2045 }
2046
2047 pub fn len(&self) -> usize {
2048 self.state.with(std::collections::HashMap::len)
2049 }
2050
2051 pub fn is_empty(&self) -> bool {
2052 self.state.with(std::collections::HashMap::is_empty)
2053 }
2054
2055 pub fn contains_key(&self, key: &K) -> bool {
2056 self.state.with(|map| map.contains_key(key))
2057 }
2058
2059 pub fn get(&self, key: &K) -> Option<V> {
2060 self.state.with(|map| map.get(key).cloned())
2061 }
2062
2063 pub fn to_hash_map(&self) -> HashMap<K, V> {
2064 self.state.with(Clone::clone)
2065 }
2066
2067 pub fn insert(&self, key: K, value: V) -> Option<V> {
2068 self.state.update(|map| map.insert(key, value))
2069 }
2070
2071 pub fn extend<I>(&self, iter: I)
2072 where
2073 I: IntoIterator<Item = (K, V)>,
2074 {
2075 self.state.update(|map| map.extend(iter));
2076 }
2077
2078 pub fn remove(&self, key: &K) -> Option<V> {
2079 self.state.update(|map| map.remove(key))
2080 }
2081
2082 pub fn clear(&self) {
2083 self.state.replace(HashMap::default());
2084 }
2085
2086 pub fn retain<F>(&self, mut predicate: F)
2087 where
2088 F: FnMut(&K, &mut V) -> bool,
2089 {
2090 self.state.update(|map| map.retain(|k, v| predicate(k, v)));
2091 }
2092}
2093
2094impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
2095where
2096 K: Clone + Eq + Hash + fmt::Debug + 'static,
2097 V: Clone + fmt::Debug + 'static,
2098{
2099 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2100 let contents = self.to_hash_map();
2101 f.debug_struct("SnapshotStateMap")
2102 .field("entries", &contents)
2103 .finish()
2104 }
2105}
2106
2107pub(crate) struct DerivedState<T: Clone + PartialEq + 'static> {
2108 compute: Rc<dyn Fn() -> T>,
2109 pub(crate) state: OwnedMutableState<T>,
2110}
2111
2112impl<T: Clone + PartialEq + 'static> DerivedState<T> {
2113 pub(crate) fn new(runtime: RuntimeHandle, compute: Rc<dyn Fn() -> T>) -> Self {
2114 let initial = compute();
2115 Self {
2116 compute,
2117 state: OwnedMutableState::with_runtime_structural_eq(initial, runtime),
2118 }
2119 }
2120
2121 pub(crate) fn set_compute(&mut self, compute: Rc<dyn Fn() -> T>) {
2122 self.compute = compute;
2123 }
2124
2125 pub(crate) fn recompute(&self) {
2126 let value = (self.compute)();
2127 self.state.set_value(value);
2128 }
2129}
2130
2131impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
2132 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2133 if let Some(value) = self.try_value() {
2134 f.debug_struct("State").field("value", &value).finish()
2135 } else {
2136 f.write_str("State { value: <unavailable> }")
2137 }
2138 }
2139}
2140
2141#[cfg(test)]
2142#[path = "tests/state_tests.rs"]
2143mod tests;