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