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 || crate::snapshot_v2::peek_next_snapshot_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
723fn clear_hidden_records(record: &StateRecord, cleared: &mut SmallVec<[Box<dyn RecordValue>; 2]>) {
729 let reuse_limit = lowest_pinned_snapshot();
730 let record_hides_all = reuse_limit.is_none_or(|limit| record.snapshot_id() < limit);
731 let mut kept: Option<Rc<StateRecord>> = None;
732 let mut cursor = record.next();
733 while let Some(node) = cursor {
734 cursor = node.next();
735 let id = node.snapshot_id();
736 if node.is_tombstone()
737 || id == PREEXISTING_SNAPSHOT_ID
738 || reuse_limit.is_some_and(|limit| id >= limit)
739 {
740 continue;
741 }
742 let hidden = if record_hides_all {
743 node
744 } else {
745 match kept.take() {
746 None => {
747 kept = Some(node);
748 continue;
749 }
750 Some(newest) if newest.snapshot_id() < id => {
751 kept = Some(node);
752 newest
753 }
754 newest => {
755 kept = newest;
756 node
757 }
758 }
759 };
760 cleared.extend(hidden.clear_value());
761 hidden.set_tombstone(true);
762 }
763}
764
765impl WriteTarget<'_> {
766 fn assert_chain_integrity(&self, caller: &str, snapshot_context: Option<SnapshotId>) {
767 assert_record_chain(self.head, self.id, caller, snapshot_context);
768 }
769
770 fn begin_global_write(&self, global: &GlobalSnapshot, snapshot_id: SnapshotId) {
771 assert!(
772 !global.has_pending_children(),
773 "SnapshotMutableState::set attempted global write while pending children {:?} exist (state {:?}, snapshot_id={})",
774 global.pending_children(),
775 self.id,
776 snapshot_id
777 );
778 mark_update_write(self.id);
779 }
780
781 fn push_global_record(&self) -> (Rc<StateRecord>, SnapshotId) {
784 let new_id = allocate_record_id();
785 (new_overwritable_record_as_head_locked(self.state), new_id)
786 }
787
788 fn publish_global_record(
791 &self,
792 record: &StateRecord,
793 new_id: SnapshotId,
794 snapshot_id: SnapshotId,
795 ) {
796 record.set_snapshot_id(new_id);
797 record.set_tombstone(false);
798 advance_global_snapshot(new_id);
799 self.assert_chain_integrity("set(global-push)", Some(snapshot_id));
800 }
801
802 fn finish_global_write(
805 &self,
806 record: &StateRecord,
807 written: Option<Rc<dyn StateObject>>,
808 snapshot_id: SnapshotId,
809 ) {
810 let mut cleared: SmallVec<[Box<dyn RecordValue>; 2]> = SmallVec::new();
811 clear_hidden_records(record, &mut cleared);
812 self.assert_chain_integrity("set(global-tombstone)", Some(snapshot_id));
813 if let Some(modified) = written.as_ref() {
814 crate::snapshot_v2::notify_apply_observers(
815 std::slice::from_ref(modified),
816 record.snapshot_id(),
817 );
818 }
819 drop(cleared);
820 }
821
822 fn begin_child_write(
823 &self,
824 snapshot: &AnySnapshot,
825 snapshot_id: SnapshotId,
826 invalid: &SnapshotIdSet,
827 written: Option<Rc<dyn StateObject>>,
828 ) -> Rc<StateRecord> {
829 if let Some(state) = written {
830 snapshot.record_write(state);
831 }
832 mark_update_write(self.id);
833 self.writable_record(snapshot_id, invalid)
834 }
835
836 fn writable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
837 let Some(readable) = readable_record_for(&self.head.clone_head(), snapshot_id, invalid)
838 else {
839 let current_head = self.head.clone_head();
840 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid);
841 let source = refreshed.unwrap_or_else(|| current_head.clone());
842
843 let cloned_value = source
844 .cloned_value()
845 .unwrap_or_else(|| panic!("StateRecord value missing or wrong type"));
846 let new_head = StateRecord::with_boxed(snapshot_id, cloned_value, Some(current_head));
847 self.head.replace(new_head.clone());
848 self.assert_chain_integrity("writable_record(recover)", Some(snapshot_id));
849 return new_head;
850 };
851
852 if readable.snapshot_id() == snapshot_id {
853 return readable;
854 }
855
856 let refreshed = {
857 let current_head = self.head.clone_head();
858 let refreshed = readable_record_for(¤t_head, snapshot_id, invalid).unwrap_or_else(
859 || {
860 panic!(
861 "SnapshotMutableState::writable_record failed to locate refreshed readable record (state {:?}, snapshot_id={}, invalid={:?})",
862 self.id, snapshot_id, invalid
863 )
864 },
865 );
866
867 if refreshed.snapshot_id() == snapshot_id {
868 return refreshed;
869 }
870
871 Rc::clone(&refreshed)
872 };
873
874 let overwritable = new_overwritable_record_locked(self.state);
875 if let Err(error) = overwritable.assign_value(&refreshed) {
876 log::error!(
877 "snapshot writable record could not copy refreshed value for state {:?}: {:?}",
878 self.id,
879 error
880 );
881 }
882 overwritable.set_snapshot_id(snapshot_id);
883 overwritable.set_tombstone(false);
884
885 self.assert_chain_integrity("writable_record(reuse)", Some(snapshot_id));
886
887 overwritable
888 }
889}
890
891#[cold]
892#[inline(never)]
893fn check_chain_integrity(
894 head: &Rc<StateRecord>,
895 id: ObjectId,
896 caller: &str,
897 snapshot_context: Option<SnapshotId>,
898) {
899 let mut cursor = Some(Rc::clone(head));
900 let mut seen: HashSet<usize> = HashSet::default();
901 let mut ids = Vec::new();
902
903 while let Some(record) = cursor {
904 let addr = Rc::as_ptr(&record) as usize;
905 assert!(
906 seen.insert(addr),
907 "SnapshotMutableState::{caller} detected duplicate/cycle at record {:p} for state {id:?} (snapshot_context={snapshot_context:?}, chain_ids={ids:?})",
908 Rc::as_ptr(&record),
909 );
910 ids.push(record.snapshot_id());
911 cursor = record.next();
912 }
913
914 assert!(
915 !ids.is_empty(),
916 "SnapshotMutableState::{caller} finished integrity scan with empty id list for state {id:?} (snapshot_context={snapshot_context:?})"
917 );
918}
919
920fn should_check_chain_integrity() -> bool {
921 #[cfg(debug_assertions)]
922 {
923 true
924 }
925
926 #[cfg(not(debug_assertions))]
927 {
928 crate::env_flag!("CRANPOSE_ASSERT_STATE_CHAIN")
929 }
930}
931
932impl<T: Clone + 'static> SnapshotMutableState<T> {
933 fn record_chain_debug(&self) -> Vec<(SnapshotId, bool)> {
934 let mut chain_ids = Vec::new();
935 let mut cursor = Some(self.first_record());
936 while let Some(record) = cursor {
937 chain_ids.push((record.snapshot_id(), record.is_tombstone()));
938 cursor = record.next();
939 }
940 chain_ids
941 }
942
943 fn readable_record_for_active_snapshot(&self) -> Result<Rc<StateRecord>, StateReadFailure> {
944 let snapshot = active_snapshot();
945 snapshot.record_read(self);
946
947 let snapshot_id = snapshot.snapshot_id();
948 if let Some(record) =
949 snapshot.with_invalid(|invalid| self.readable_for(snapshot_id, invalid))
950 {
951 return Ok(record);
952 }
953
954 let fresh_snapshot = active_snapshot();
955 let fresh_id = fresh_snapshot.snapshot_id();
956 let fresh_invalid = fresh_snapshot.invalid();
957
958 if let Some(record) = self.readable_for(fresh_id, &fresh_invalid) {
959 return Ok(record);
960 }
961
962 let global = GlobalSnapshot::get_or_create();
963 let global_id = global.snapshot_id();
964 let global_invalid = global.invalid();
965
966 if let Some(record) = self.readable_for(global_id, &global_invalid) {
967 return Ok(record);
968 }
969
970 Err(StateReadFailure {
971 state_id: self.id,
972 snapshot_id,
973 fresh_snapshot_id: fresh_id,
974 fresh_invalid,
975 record_chain: self.record_chain_debug(),
976 })
977 }
978
979 fn readable_for(
980 &self,
981 snapshot_id: SnapshotId,
982 invalid: &SnapshotIdSet,
983 ) -> Option<Rc<StateRecord>> {
984 let head = self.first_record();
985 readable_record_for(&head, snapshot_id, invalid)
986 }
987
988 fn is_equivalent_to(&self, readable: Option<&StateRecord>, new_value: &T) -> bool {
989 readable.is_some_and(|record| {
990 record.with_value(|current: &T| self.policy.equivalent(current, new_value))
991 })
992 }
993
994 pub(crate) fn new_in_rc(initial: T, policy: Rc<dyn MutationPolicy<T>>) -> Rc<Self> {
995 let snapshot = active_snapshot();
996 let snapshot_id = snapshot.snapshot_id();
997
998 let tail = StateRecord::new(PREEXISTING_SNAPSHOT_ID, initial.clone(), None);
999 let head = StateRecord::new(snapshot_id, initial, Some(tail));
1000
1001 Rc::new_cyclic(|weak_self| Self {
1002 head: CurrentRecord::new(head),
1003 policy,
1004 id: ObjectId(weak_self.as_ptr() as usize),
1005 weak_self: weak_self.clone(),
1006 apply_observer: OnceCell::new(),
1007 read_observation_lease: Rc::new(()),
1008 scope_observation_count: Cell::new(0),
1009 subscriber_callbacks: RefCell::new(Vec::new()),
1010 })
1011 }
1012
1013 pub(crate) fn set_apply_observer(&self, observer: Box<dyn Fn() + 'static>) {
1014 assert!(
1015 self.apply_observer.set(observer).is_ok(),
1016 "state observer already installed"
1017 );
1018 }
1019
1020 fn acquire_observation_lease(&self) -> Option<Rc<dyn Any>> {
1021 let was_empty = !self.has_subscribers();
1022 let lease = Rc::clone(&self.read_observation_lease);
1023 if was_empty {
1024 notify_subscriber_callbacks(&self.subscriber_callbacks);
1025 }
1026 Some(lease)
1027 }
1028
1029 fn add_scope_observer(&self) -> bool {
1030 let was_empty = !self.has_subscribers();
1031 let count = self
1032 .scope_observation_count
1033 .get()
1034 .checked_add(1)
1035 .expect("state scope observation count overflow");
1036 self.scope_observation_count.set(count);
1037 was_empty
1038 }
1039
1040 fn remove_scope_observers(&self, count: usize) {
1041 if count == 0 {
1042 return;
1043 }
1044 let remaining = self
1045 .scope_observation_count
1046 .get()
1047 .checked_sub(count)
1048 .expect("state scope observation count underflow");
1049 self.scope_observation_count.set(remaining);
1050 }
1051
1052 fn has_subscribers(&self) -> bool {
1053 Rc::strong_count(&self.read_observation_lease) > 1 || self.scope_observation_count.get() > 0
1054 }
1055
1056 fn subscriber_callback(&self, callback: Rc<dyn Fn()>, notify: bool) {
1057 self.subscriber_callbacks
1058 .borrow_mut()
1059 .push(Rc::clone(&callback));
1060 if notify {
1061 callback();
1062 }
1063 }
1064
1065 fn notify_subscribers(&self) {
1066 notify_subscriber_callbacks(&self.subscriber_callbacks);
1067 }
1068
1069 fn finish_apply(&self, caller: &str, snapshot_id: SnapshotId) {
1070 finish_apply(
1071 &self.apply_observer,
1072 &self.head,
1073 self.id,
1074 caller,
1075 snapshot_id,
1076 );
1077 }
1078
1079 fn upgrade_self(&self) -> Option<Rc<Self>> {
1080 self.weak_self.upgrade()
1081 }
1082
1083 fn written_state(&self) -> Option<Rc<dyn StateObject>> {
1084 self.upgrade_self()
1085 .map(|state| -> Rc<dyn StateObject> { state })
1086 }
1087
1088 #[inline]
1089 pub(crate) fn id(&self) -> ObjectId {
1090 self.id
1091 }
1092
1093 pub(crate) fn try_with_value<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1094 let record = self.readable_record_for_active_snapshot().ok()?;
1095 record.try_with_value(f)
1096 }
1097
1098 pub(crate) fn try_get(&self) -> Option<T> {
1099 self.try_with_value(Clone::clone)
1100 }
1101
1102 pub(crate) fn with_value<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1103 let record = self
1104 .readable_record_for_active_snapshot()
1105 .unwrap_or_else(|failure| panic!("{failure}"));
1106 record.with_value(f)
1107 }
1108
1109 pub(crate) fn get(&self) -> T {
1110 self.with_value(Clone::clone)
1111 }
1112
1113 pub(crate) fn set(&self, new_value: T) -> bool {
1114 #[cfg(debug_assertions)]
1115 {
1116 let in_handler = crate::in_event_handler();
1117 let in_snapshot = crate::in_applied_snapshot();
1118 if in_handler && !in_snapshot {
1119 log::warn!(
1120 target: "cranpose::state",
1121 "State modified in event handler without run_in_mutable_snapshot; \
1122 this can make updates invisible to other contexts. Wrap the handler \
1123 in run_in_mutable_snapshot() or dispatch_ui_event(). State: {:?}",
1124 self.id
1125 );
1126 }
1127 }
1128
1129 let snapshot = active_snapshot();
1130 let snapshot_id = snapshot.snapshot_id();
1131 let target = WriteTarget {
1132 head: &self.head,
1133 id: self.id,
1134 state: self,
1135 };
1136
1137 match &snapshot {
1138 AnySnapshot::Global(global) => {
1139 let readable =
1140 snapshot.with_invalid(|invalid| self.readable_for(snapshot_id, invalid));
1141 if self.is_equivalent_to(readable.as_deref(), &new_value) {
1142 return false;
1143 }
1144 target.begin_global_write(global, snapshot_id);
1145 if let Some(record) = readable.filter(|record| record.snapshot_id() == snapshot_id)
1150 {
1151 let displaced = record.replace_value(new_value);
1152 target.finish_global_write(&record, self.written_state(), snapshot_id);
1153 drop(displaced);
1154 } else {
1155 let (record, new_id) = target.push_global_record();
1156 let displaced = record.replace_value(new_value);
1157 target.publish_global_record(&record, new_id, snapshot_id);
1158 target.finish_global_write(&record, self.written_state(), snapshot_id);
1159 drop(displaced);
1160 }
1161 }
1162 AnySnapshot::Mutable(_)
1163 | AnySnapshot::NestedMutable(_)
1164 | AnySnapshot::TransparentMutable(_) => {
1165 let invalid = snapshot.invalid();
1166 if self.is_equivalent_to(
1167 self.readable_for(snapshot_id, &invalid).as_deref(),
1168 &new_value,
1169 ) {
1170 return false;
1171 }
1172 let record = target.begin_child_write(
1173 &snapshot,
1174 snapshot_id,
1175 &invalid,
1176 self.written_state(),
1177 );
1178 let displaced = record.replace_value(new_value);
1179 target.assert_chain_integrity("set(child-writable)", Some(snapshot_id));
1180 drop(displaced);
1181 }
1182 AnySnapshot::Readonly(_)
1183 | AnySnapshot::NestedReadonly(_)
1184 | AnySnapshot::TransparentReadonly(_) => {
1185 panic!("Cannot write to a read-only snapshot");
1186 }
1187 }
1188
1189 true
1190 }
1191}
1192
1193pub(crate) struct UpdateScope {
1194 id: ObjectId,
1195 finished: bool,
1196}
1197
1198impl UpdateScope {
1199 pub(crate) fn new(id: ObjectId) -> Self {
1200 crate::thread_state::snapshots(|state| {
1201 state.active_updates.borrow_mut().insert(id);
1202 state.pending_writes.borrow_mut().remove(&id);
1203 });
1204 Self {
1205 id,
1206 finished: false,
1207 }
1208 }
1209
1210 pub(crate) fn finish(mut self) -> bool {
1211 self.finished = true;
1212 crate::thread_state::snapshots(|state| {
1213 state.active_updates.borrow_mut().remove(&self.id);
1214 state.pending_writes.borrow_mut().remove(&self.id)
1215 })
1216 }
1217}
1218
1219impl Drop for UpdateScope {
1220 fn drop(&mut self) {
1221 if self.finished {
1222 return;
1223 }
1224 crate::thread_state::snapshots(|state| {
1225 state.active_updates.borrow_mut().remove(&self.id);
1226 state.pending_writes.borrow_mut().remove(&self.id);
1227 });
1228 }
1229}
1230
1231fn mark_update_write(id: ObjectId) {
1232 crate::thread_state::snapshots(|state| {
1233 if state.active_updates.borrow().contains(&id) {
1234 state.pending_writes.borrow_mut().insert(id);
1235 }
1236 });
1237}
1238
1239impl<T: Clone + 'static> SnapshotMutableState<T> {
1240 fn try_readable_record(
1241 &self,
1242 snapshot_id: SnapshotId,
1243 invalid: &SnapshotIdSet,
1244 ) -> Option<Rc<StateRecord>> {
1245 self.readable_for(snapshot_id, invalid)
1246 }
1247}
1248
1249impl<T: Clone + 'static> StateObject for SnapshotMutableState<T> {
1250 fn object_id(&self) -> ObjectId {
1251 self.id
1252 }
1253
1254 fn first_record(&self) -> Rc<StateRecord> {
1255 self.head.clone_head()
1256 }
1257
1258 fn try_readable_record(
1259 &self,
1260 snapshot_id: SnapshotId,
1261 invalid: &SnapshotIdSet,
1262 ) -> Option<Rc<StateRecord>> {
1263 self.try_readable_record(snapshot_id, invalid)
1264 }
1265
1266 fn readable_record(&self, snapshot_id: SnapshotId, invalid: &SnapshotIdSet) -> Rc<StateRecord> {
1267 self.try_readable_record(snapshot_id, invalid)
1268 .unwrap_or_else(|| {
1269 panic!(
1270 "SnapshotMutableState::readable_record returned null (state={:?}, snapshot_id={})",
1271 self.id, snapshot_id
1272 )
1273 })
1274 }
1275
1276 fn prepend_state_record(&self, record: Rc<StateRecord>) {
1277 self.head.prepend(record);
1278 }
1279
1280 fn observation_lease(&self) -> Option<Rc<dyn Any>> {
1281 self.acquire_observation_lease()
1282 }
1283
1284 fn merge_records(
1285 &self,
1286 previous: Rc<StateRecord>,
1287 current: Rc<StateRecord>,
1288 applied: Rc<StateRecord>,
1289 ) -> Option<Rc<StateRecord>> {
1290 let outcome = current
1291 .try_with_value(|current_value: &T| {
1292 applied
1293 .try_with_value(|applied_value: &T| {
1294 if self.policy.equivalent(current_value, applied_value) {
1295 return Merge::KeepCurrent;
1296 }
1297 previous
1298 .try_with_value(|previous_value: &T| {
1299 Merge::Value(self.policy.merge(
1300 previous_value,
1301 current_value,
1302 applied_value,
1303 ))
1304 })
1305 .unwrap_or(Merge::Missing("previous", &previous))
1306 })
1307 .unwrap_or(Merge::Missing("applied", &applied))
1308 })
1309 .unwrap_or(Merge::Missing("current", ¤t));
1310 match outcome {
1311 Merge::KeepCurrent => Some(current),
1312 Merge::Value(merged) => {
1313 merged.map(|value| StateRecord::new(applied.snapshot_id(), value, None))
1314 }
1315 Merge::Missing(which, record) => {
1316 log_merge_value_missing(which, self.id, record);
1317 None
1318 }
1319 }
1320 }
1321
1322 fn promote_record(&self, child_id: SnapshotId) -> Result<(), &'static str> {
1323 promote_record_locked(self, child_id, TypeId::of::<T>())?;
1324 self.finish_apply("promote_record", child_id);
1325 Ok(())
1326 }
1327
1328 fn commit_merged_record(&self, merged: Rc<StateRecord>) -> Result<SnapshotId, &'static str> {
1329 let new_id = commit_merged_record_locked(self, &merged, TypeId::of::<T>())?;
1330 self.finish_apply("commit_merged_record", new_id);
1331 Ok(new_id)
1332 }
1333
1334 fn overwrite_unused_records(&self) -> bool {
1335 overwrite_unused_records_locked(self)
1336 }
1337
1338 fn as_any(&self) -> &dyn Any {
1339 self
1340 }
1341}
1342
1343pub(crate) struct MutableStateInner<T: Clone + 'static> {
1344 pub(crate) state: Rc<SnapshotMutableState<T>>,
1345 pub(crate) watchers: RefCell<HashMap<ScopeId, Weak<RecomposeScopeInner>>>,
1346 runtime: RuntimeHandle,
1347 state_id: Cell<Option<StateId>>,
1348}
1349
1350fn notify_subscriber_callbacks(callbacks: &RefCell<Vec<Rc<dyn Fn()>>>) {
1351 let registered = callbacks.borrow().len();
1352 for index in 0..registered {
1353 let callback = callbacks.borrow().get(index).map(Rc::clone);
1354 if let Some(callback) = callback {
1355 callback();
1356 }
1357 }
1358}
1359
1360fn shrink_watchers_if_sparse(watchers: &mut HashMap<ScopeId, Weak<RecomposeScopeInner>>) {
1361 let len = watchers.len();
1362 let capacity = watchers.capacity();
1363 if capacity > len.saturating_mul(4).max(32) {
1364 watchers.shrink_to_fit();
1365 }
1366}
1367
1368impl<T: Clone + 'static> MutableStateInner<T> {
1369 pub(crate) fn new_with_policy(
1370 value: T,
1371 runtime: RuntimeHandle,
1372 policy: Rc<dyn MutationPolicy<T>>,
1373 ) -> Self {
1374 Self {
1375 state: SnapshotMutableState::new_in_rc(value, policy),
1376 watchers: RefCell::new(HashMap::default()),
1377 runtime,
1378 state_id: Cell::new(None),
1379 }
1380 }
1381
1382 pub(crate) fn install_snapshot_observer(&self, state_id: StateId) {
1383 self.state_id.set(Some(state_id));
1384 let runtime_handle = self.runtime.clone();
1385 self.state.set_apply_observer(Box::new(move || {
1386 let runtime = runtime_handle.clone();
1387 runtime_handle.enqueue_ui_task(Box::new(move || {
1388 runtime.with_state_arena(|arena| {
1389 let _ = arena.with_typed_opt::<T, _>(state_id, |inner| {
1390 inner.invalidate_watchers();
1391 });
1392 });
1393 }));
1394 }));
1395 }
1396
1397 fn register_scope(&self, scope: &RecomposeScope) -> (bool, bool) {
1407 let mut watchers = self.watchers.borrow_mut();
1408 let id = scope.id();
1409 match watchers.get(&id) {
1410 Some(existing) if existing.strong_count() > 0 => return (false, false),
1411 Some(_) => {
1412 watchers.remove(&id);
1413 self.state.remove_scope_observers(1);
1414 }
1415 None => {}
1416 }
1417 let any_live = watchers.values().any(|watcher| watcher.strong_count() > 0);
1418 if !any_live || watchers.len() >= watchers.capacity() {
1419 let before = watchers.len();
1420 watchers.retain(|_, watcher| watcher.strong_count() > 0);
1421 self.state.remove_scope_observers(before - watchers.len());
1422 }
1423 watchers.insert(id, scope.downgrade());
1424 drop(watchers);
1425 (true, self.state.add_scope_observer())
1426 }
1427
1428 fn has_subscribers(&self) -> bool {
1429 let mut watchers = self.watchers.borrow_mut();
1430 let before = watchers.len();
1431 watchers.retain(|_, existing| existing.upgrade().is_some());
1432 self.state.remove_scope_observers(before - watchers.len());
1433 self.state.has_subscribers()
1434 }
1435
1436 pub(crate) fn forget_scope(&self, scope_id: ScopeId) {
1439 let mut watchers = self.watchers.borrow_mut();
1440 let removed = watchers.remove(&scope_id).is_some();
1441 if removed {
1442 shrink_watchers_if_sparse(&mut watchers);
1443 }
1444 drop(watchers);
1445 self.state.remove_scope_observers(usize::from(removed));
1446 }
1447
1448 pub(crate) fn unregister_scope(&self, scope_id: ScopeId) {
1449 let mut watchers = self.watchers.borrow_mut();
1450 let removed = if watchers
1451 .get(&scope_id)
1452 .is_some_and(|weak| weak.upgrade().is_none())
1453 {
1454 watchers.remove(&scope_id);
1455 shrink_watchers_if_sparse(&mut watchers);
1456 true
1457 } else {
1458 false
1459 };
1460 drop(watchers);
1461 self.state.remove_scope_observers(usize::from(removed));
1462 }
1463
1464 fn state_id(&self) -> Option<StateId> {
1465 self.state_id.get()
1466 }
1467
1468 fn invalidate_watchers(&self) {
1469 let (watchers, removed_count): (smallvec::SmallVec<[RecomposeScope; 4]>, usize) = {
1471 let mut watchers = self.watchers.borrow_mut();
1472 let before = watchers.len();
1473 let mut live = smallvec::SmallVec::new();
1474 watchers.retain(|_, scope| {
1475 if let Some(inner) = scope.upgrade() {
1476 live.push(RecomposeScope { inner });
1477 true
1478 } else {
1479 false
1480 }
1481 });
1482 let removed_count = before - watchers.len();
1483 shrink_watchers_if_sparse(&mut watchers);
1484 (live, removed_count)
1485 };
1486 self.state.remove_scope_observers(removed_count);
1487
1488 for watcher in watchers {
1489 debug_record_scope_invalidation(
1490 watcher.id(),
1491 self.state_id.get(),
1492 std::any::type_name::<T>(),
1493 );
1494 if let Some(state_id) = self.state_id.get() {
1495 watcher.invalidate_from_state(state_id);
1496 } else {
1497 watcher.invalidate();
1498 }
1499 }
1500 }
1501}
1502
1503impl<T: Clone + 'static> Drop for MutableStateInner<T> {
1504 fn drop(&mut self) {
1505 self.state
1506 .remove_scope_observers(self.watchers.get_mut().len());
1507 }
1508}
1509
1510fn register_current_state_scope<T: Clone + 'static>(inner: &MutableStateInner<T>) {
1511 let state_id = inner.state_id();
1512 let Some(Some((scope, read))) =
1513 with_current_composer_opt(|composer| composer.note_state_read(state_id))
1514 else {
1515 return;
1516 };
1517 let (registered, became_subscribed) = inner.register_scope(&scope);
1518 if let (crate::ScopeRead::NotReading, true, Some(state_id)) = (read, registered, state_id) {
1519 scope.record_state_subscription(state_id);
1520 }
1521 if became_subscribed {
1522 inner.state.notify_subscribers();
1523 }
1524}
1525
1526trait StateArenaHandle<T: Clone + 'static> {
1527 fn state_id(&self) -> StateId;
1528 fn runtime_id(&self) -> runtime::RuntimeId;
1529
1530 fn runtime_handle(&self) -> RuntimeHandle {
1531 runtime::runtime_handle_by_id(self.runtime_id())
1532 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1533 }
1534
1535 fn runtime_handle_opt(&self) -> Option<RuntimeHandle> {
1536 runtime::runtime_handle_by_id(self.runtime_id())
1537 }
1538
1539 fn with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1540 runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1541 arena.with_typed::<T, R>(self.state_id(), f)
1542 })
1543 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id()))
1544 }
1545
1546 fn try_with_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1547 runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1548 arena.with_typed_opt::<T, R>(self.state_id(), f)
1549 })?
1550 }
1551
1552 fn read_subscribed<R>(&self, read: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1553 self.with_inner(|inner| {
1554 let result = read(inner);
1555 register_current_state_scope(inner);
1556 result
1557 })
1558 }
1559}
1560
1561pub struct State<T: Clone + 'static> {
1563 id: StateId,
1564 runtime_id: runtime::RuntimeId,
1565 derived: bool,
1568 _marker: PhantomData<fn() -> T>,
1569}
1570
1571pub struct MutableState<T: Clone + 'static> {
1577 id: StateId,
1578 runtime_id: runtime::RuntimeId,
1579 _marker: PhantomData<fn() -> T>,
1580}
1581
1582#[derive(Clone)]
1584pub struct OwnedMutableState<T: Clone + 'static> {
1585 state: MutableState<T>,
1586 _lease: Rc<runtime::StateHandleLease>,
1587 _marker: PhantomData<fn() -> T>,
1588}
1589
1590impl<T: Clone + 'static> PartialEq for State<T> {
1591 fn eq(&self, other: &Self) -> bool {
1592 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1593 }
1594}
1595
1596impl<T: Clone + 'static> Eq for State<T> {}
1597
1598impl<T: Clone + 'static> PartialEq for MutableState<T> {
1599 fn eq(&self, other: &Self) -> bool {
1600 self.state_id() == other.state_id() && self.runtime_id() == other.runtime_id()
1601 }
1602}
1603
1604impl<T: Clone + 'static> PartialEq for OwnedMutableState<T> {
1607 fn eq(&self, other: &Self) -> bool {
1608 self.state == other.state
1609 }
1610}
1611
1612impl<T: Clone + 'static> Eq for MutableState<T> {}
1613
1614impl<T: Clone + 'static> Copy for State<T> {}
1615
1616impl<T: Clone + 'static> Clone for State<T> {
1617 fn clone(&self) -> Self {
1618 *self
1619 }
1620}
1621
1622impl<T: Clone + 'static> Copy for MutableState<T> {}
1623
1624impl<T: Clone + 'static> Clone for MutableState<T> {
1625 fn clone(&self) -> Self {
1626 *self
1627 }
1628}
1629
1630impl<T: Clone + 'static> StateArenaHandle<T> for State<T> {
1631 fn state_id(&self) -> StateId {
1632 self.id
1633 }
1634
1635 fn runtime_id(&self) -> runtime::RuntimeId {
1636 self.runtime_id
1637 }
1638}
1639
1640impl<T: Clone + 'static> State<T> {
1641 pub fn is_alive(&self) -> bool {
1644 self.try_with_inner(|_| ()).is_some()
1645 }
1646
1647 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1650 self.try_with_fresh_inner(|inner| inner.state.try_with_value(f))?
1651 }
1652
1653 pub fn try_value(&self) -> Option<T> {
1656 self.try_with_fresh_inner(|inner| inner.state.try_get())?
1657 }
1658
1659 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1663 let value = self.read_fresh_subscribed(|inner| inner.state.get());
1664 f(&value)
1665 }
1666
1667 pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1672 self.read_fresh_subscribed(|inner| inner.state.with_value(f))
1673 }
1674
1675 pub fn value(&self) -> T {
1678 self.read_fresh_subscribed(|inner| inner.state.get())
1679 }
1680
1681 pub fn get(&self) -> T {
1683 self.value()
1684 }
1685
1686 pub fn get_non_reactive(&self) -> T {
1690 self.with_fresh_inner(|inner| inner.state.get())
1691 }
1692
1693 fn of_derived(self) -> Self {
1695 Self {
1696 derived: true,
1697 ..self
1698 }
1699 }
1700
1701 fn with_fresh_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1706 runtime::with_state_arena_by_id(self.runtime_id, |arena| {
1707 if self.derived {
1708 arena.refresh_derived(self.id);
1709 }
1710 arena.with_typed::<T, R>(self.id, f)
1711 })
1712 .unwrap_or_else(|| panic!("runtime {:?} dropped", self.runtime_id))
1713 }
1714
1715 fn try_with_fresh_inner<R>(&self, f: impl FnOnce(&MutableStateInner<T>) -> R) -> Option<R> {
1717 runtime::with_state_arena_by_id(self.runtime_id, |arena| {
1718 if self.derived {
1719 arena.refresh_derived(self.id);
1720 }
1721 arena.with_typed_opt::<T, R>(self.id, f)
1722 })?
1723 }
1724
1725 fn read_fresh_subscribed<R>(&self, read: impl FnOnce(&MutableStateInner<T>) -> R) -> R {
1728 self.with_fresh_inner(|inner| {
1729 let result = read(inner);
1730 register_current_state_scope(inner);
1731 result
1732 })
1733 }
1734
1735 pub fn has_subscribers(&self) -> bool {
1737 self.with_inner(MutableStateInner::has_subscribers)
1738 }
1739
1740 pub fn on_subscriber(&self, callback: impl Fn() + 'static) {
1746 let callback: Rc<dyn Fn()> = Rc::new(callback);
1747 self.with_inner(|inner| {
1748 inner
1749 .state
1750 .subscriber_callback(callback, inner.has_subscribers());
1751 });
1752 }
1753
1754 pub fn subscription_hold(&self) -> StateSubscriptionHold {
1766 StateSubscriptionHold {
1767 _lease: self
1768 .try_with_inner(|inner| inner.state.observation_lease())
1769 .flatten(),
1770 }
1771 }
1772}
1773
1774pub struct StateSubscriptionHold {
1777 _lease: Option<Rc<dyn Any>>,
1778}
1779
1780impl<T: Clone + 'static> StateArenaHandle<T> for MutableState<T> {
1781 fn state_id(&self) -> StateId {
1782 self.id
1783 }
1784
1785 fn runtime_id(&self) -> runtime::RuntimeId {
1786 self.runtime_id
1787 }
1788}
1789
1790impl<T: Clone + 'static> MutableState<T> {
1791 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1794 runtime.alloc_persistent_state(value)
1795 }
1796
1797 fn from_parts(id: StateId, runtime_id: runtime::RuntimeId) -> Self {
1798 Self {
1799 id,
1800 runtime_id,
1801 _marker: PhantomData,
1802 }
1803 }
1804
1805 pub(crate) fn from_lease(lease: &Rc<runtime::StateHandleLease>) -> Self {
1806 Self::from_parts(lease.id(), lease.runtime_id())
1807 }
1808
1809 pub fn is_alive(&self) -> bool {
1812 self.try_with_inner(|_| ()).is_some()
1813 }
1814
1815 pub fn try_with<R>(&self, f: impl FnOnce(&T) -> R) -> Option<R> {
1818 self.try_with_inner(|inner| inner.state.try_with_value(f))?
1819 }
1820
1821 pub fn try_value(&self) -> Option<T> {
1824 self.try_with_inner(|inner| inner.state.try_get())?
1825 }
1826
1827 pub fn as_state(&self) -> State<T> {
1829 State {
1830 id: self.id,
1831 runtime_id: self.runtime_id,
1832 derived: false,
1833 _marker: PhantomData,
1834 }
1835 }
1836
1837 pub fn try_retain(&self) -> Option<OwnedMutableState<T>> {
1840 let lease = self
1841 .runtime_handle_opt()?
1842 .retain_state_lease(self.state_id())?;
1843 Some(OwnedMutableState {
1844 state: *self,
1845 _lease: lease,
1846 _marker: PhantomData,
1847 })
1848 }
1849
1850 pub fn retain(&self) -> OwnedMutableState<T> {
1853 self.try_retain()
1854 .unwrap_or_else(|| panic!("state {:?} is no longer alive", self.state_id()))
1855 }
1856
1857 pub fn with<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1861 let value = self.read_subscribed(|inner| inner.state.get());
1862 f(&value)
1863 }
1864
1865 pub fn read<R>(&self, f: impl FnOnce(&T) -> R) -> R {
1870 self.read_subscribed(|inner| inner.state.with_value(f))
1871 }
1872
1873 pub fn update<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
1876 let runtime = self.runtime_handle();
1877 runtime.assert_ui_thread();
1878 runtime.with_state_arena(|arena| {
1879 arena.with_typed::<T, R>(self.state_id(), |inner| {
1880 let mut value = inner.state.get();
1881 let tracker = UpdateScope::new(inner.state.id());
1882 let result = f(&mut value);
1883 let wrote_elsewhere = tracker.finish();
1884 if !wrote_elsewhere && inner.state.set(value) {
1885 inner.invalidate_watchers();
1886 }
1887 result
1888 })
1889 })
1890 }
1891
1892 pub fn replace(&self, value: T) {
1895 let replaced = runtime::with_state_arena_by_id(self.runtime_id(), |arena| {
1898 arena.with_typed_opt::<T, ()>(self.state_id(), |inner| {
1899 if inner.state.set(value) {
1900 inner.invalidate_watchers();
1901 }
1902 })
1903 });
1904 match replaced {
1905 None => log::debug!(
1906 "MutableState::replace skipped: runtime {:?} dropped",
1907 self.runtime_id()
1908 ),
1909 Some(None) => log::debug!(
1910 "MutableState::replace skipped: state cell released (slot={}, gen={})",
1911 self.state_id().slot(),
1912 self.state_id().generation(),
1913 ),
1914 Some(Some(())) => {}
1915 }
1916 }
1917
1918 pub fn set_value(&self, value: T) {
1920 self.replace(value);
1921 }
1922
1923 pub fn set(&self, value: T) {
1925 self.replace(value);
1926 }
1927
1928 pub fn value(&self) -> T {
1931 self.read_subscribed(|inner| inner.state.get())
1932 }
1933
1934 pub fn get(&self) -> T {
1936 self.value()
1937 }
1938
1939 pub fn get_non_reactive(&self) -> T {
1941 self.with_inner(|inner| inner.state.get())
1942 }
1943
1944 #[doc(hidden)]
1945 pub fn runtime_state_id(&self) -> StateId {
1946 self.state_id()
1947 }
1948
1949 #[doc(hidden)]
1950 pub fn subscribe_current_scope_only(&self) {
1951 self.subscribe_current_scope();
1952 }
1953
1954 fn subscribe_current_scope(&self) {
1955 self.with_inner(register_current_state_scope::<T>);
1956 }
1957
1958 #[cfg(test)]
1959 pub(crate) fn watcher_count(&self) -> usize {
1960 self.with_inner(|inner| inner.watchers.borrow().len())
1961 }
1962
1963 #[cfg(test)]
1964 pub(crate) fn watcher_capacity(&self) -> usize {
1965 self.with_inner(|inner| inner.watchers.borrow().capacity())
1966 }
1967
1968 #[cfg(test)]
1969 pub(crate) fn state_id_for_test(&self) -> StateId {
1970 self.state_id()
1971 }
1972
1973 #[cfg(test)]
1974 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
1975 self.as_state().subscribe_scope_for_test(scope);
1976 }
1977}
1978
1979impl<T: Clone + 'static> OwnedMutableState<T> {
1980 pub fn with_runtime(value: T, runtime: RuntimeHandle) -> Self {
1983 let lease = runtime.alloc_state(value);
1984 Self {
1985 state: MutableState::from_lease(&lease),
1986 _lease: lease,
1987 _marker: PhantomData,
1988 }
1989 }
1990
1991 pub fn with_runtime_structural_eq(value: T, runtime: RuntimeHandle) -> Self
1994 where
1995 T: PartialEq,
1996 {
1997 Self::with_runtime_and_policy(value, runtime, Rc::new(StructuralEqual))
1998 }
1999
2000 pub(crate) fn with_runtime_and_policy(
2001 value: T,
2002 runtime: RuntimeHandle,
2003 policy: Rc<dyn MutationPolicy<T>>,
2004 ) -> Self {
2005 let lease = runtime.alloc_state_with_policy(value, policy);
2006 Self {
2007 state: MutableState::from_lease(&lease),
2008 _lease: lease,
2009 _marker: PhantomData,
2010 }
2011 }
2012
2013 pub fn handle(&self) -> MutableState<T> {
2015 self.state
2016 }
2017
2018 pub fn as_state(&self) -> State<T> {
2020 self.state.as_state()
2021 }
2022}
2023
2024impl<T: Clone + 'static> Deref for OwnedMutableState<T> {
2025 type Target = MutableState<T>;
2026
2027 fn deref(&self) -> &Self::Target {
2028 &self.state
2029 }
2030}
2031
2032#[cfg(test)]
2033impl<T: Clone + 'static> State<T> {
2034 pub(crate) fn subscribe_scope_for_test(&self, scope: &RecomposeScope) {
2035 self.with_inner(|inner| {
2036 let (registered, became_subscribed) = inner.register_scope(scope);
2037 if registered {
2038 if let Some(state_id) = inner.state_id() {
2039 scope.record_state_subscription(state_id);
2040 }
2041 if became_subscribed {
2042 inner.state.notify_subscribers();
2043 }
2044 }
2045 });
2046 }
2047}
2048
2049impl<T: fmt::Debug + Clone + 'static> fmt::Debug for MutableState<T> {
2050 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2051 if let Some(value) = self.try_value() {
2052 f.debug_struct("MutableState")
2053 .field("value", &value)
2054 .finish()
2055 } else {
2056 f.write_str("MutableState { value: <unavailable> }")
2057 }
2058 }
2059}
2060
2061#[derive(Clone)]
2064pub struct SnapshotStateList<T: Clone + 'static> {
2065 state: OwnedMutableState<Vec<T>>,
2066}
2067
2068impl<T: Clone + 'static> SnapshotStateList<T> {
2069 pub fn with_runtime<I>(values: I, runtime: RuntimeHandle) -> Self
2071 where
2072 I: IntoIterator<Item = T>,
2073 {
2074 let initial: Vec<T> = values.into_iter().collect();
2075 Self {
2076 state: OwnedMutableState::with_runtime(initial, runtime),
2077 }
2078 }
2079
2080 pub fn as_state(&self) -> State<Vec<T>> {
2082 self.state.as_state()
2083 }
2084
2085 pub fn as_mutable_state(&self) -> MutableState<Vec<T>> {
2087 self.state.handle()
2088 }
2089
2090 pub fn len(&self) -> usize {
2092 self.state.with(Vec::len)
2093 }
2094
2095 pub fn is_empty(&self) -> bool {
2097 self.len() == 0
2098 }
2099
2100 pub fn to_vec(&self) -> Vec<T> {
2102 self.state.with(Clone::clone)
2103 }
2104
2105 pub fn iter(&self) -> Vec<T> {
2107 self.to_vec()
2108 }
2109
2110 pub fn get(&self, index: usize) -> T {
2112 self.state.with(|values| values[index].clone())
2113 }
2114
2115 pub fn get_opt(&self, index: usize) -> Option<T> {
2117 self.state.with(|values| values.get(index).cloned())
2118 }
2119
2120 pub fn first(&self) -> Option<T> {
2122 self.get_opt(0)
2123 }
2124
2125 pub fn last(&self) -> Option<T> {
2127 self.state.with(|values| values.last().cloned())
2128 }
2129
2130 pub fn push(&self, value: T) {
2132 self.state.update(|values| values.push(value));
2133 }
2134
2135 pub fn extend<I>(&self, iter: I)
2137 where
2138 I: IntoIterator<Item = T>,
2139 {
2140 self.state.update(|values| values.extend(iter));
2141 }
2142
2143 pub fn insert(&self, index: usize, value: T) {
2145 self.state.update(|values| values.insert(index, value));
2146 }
2147
2148 pub fn set(&self, index: usize, value: T) -> T {
2150 self.state
2151 .update(|values| std::mem::replace(&mut values[index], value))
2152 }
2153
2154 pub fn remove(&self, index: usize) -> T {
2156 self.state.update(|values| values.remove(index))
2157 }
2158
2159 pub fn pop(&self) -> Option<T> {
2161 self.state.update(Vec::pop)
2162 }
2163
2164 pub fn clear(&self) {
2166 self.state.replace(Vec::new());
2167 }
2168
2169 pub fn retain<F>(&self, mut predicate: F)
2171 where
2172 F: FnMut(&T) -> bool,
2173 {
2174 self.state
2175 .update(|values| values.retain(|value| predicate(value)));
2176 }
2177
2178 pub fn replace_with<I>(&self, iter: I)
2180 where
2181 I: IntoIterator<Item = T>,
2182 {
2183 self.state.replace(iter.into_iter().collect());
2184 }
2185}
2186
2187impl<T: fmt::Debug + Clone + 'static> fmt::Debug for SnapshotStateList<T> {
2188 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2189 let contents = self.to_vec();
2190 f.debug_struct("SnapshotStateList")
2191 .field("values", &contents)
2192 .finish()
2193 }
2194}
2195
2196#[derive(Clone)]
2199pub struct SnapshotStateMap<K, V>
2200where
2201 K: Clone + Eq + Hash + 'static,
2202 V: Clone + 'static,
2203{
2204 state: OwnedMutableState<HashMap<K, V>>,
2205}
2206
2207impl<K, V> SnapshotStateMap<K, V>
2208where
2209 K: Clone + Eq + Hash + 'static,
2210 V: Clone + 'static,
2211{
2212 pub fn with_runtime<I>(pairs: I, runtime: RuntimeHandle) -> Self
2214 where
2215 I: IntoIterator<Item = (K, V)>,
2216 {
2217 let map: HashMap<K, V> = pairs.into_iter().collect();
2218 Self {
2219 state: OwnedMutableState::with_runtime(map, runtime),
2220 }
2221 }
2222
2223 pub fn as_state(&self) -> State<HashMap<K, V>> {
2225 self.state.as_state()
2226 }
2227
2228 pub fn as_mutable_state(&self) -> MutableState<HashMap<K, V>> {
2230 self.state.handle()
2231 }
2232
2233 pub fn len(&self) -> usize {
2235 self.state.with(std::collections::HashMap::len)
2236 }
2237
2238 pub fn is_empty(&self) -> bool {
2240 self.state.with(std::collections::HashMap::is_empty)
2241 }
2242
2243 pub fn contains_key(&self, key: &K) -> bool {
2245 self.state.with(|map| map.contains_key(key))
2246 }
2247
2248 pub fn get(&self, key: &K) -> Option<V> {
2250 self.state.with(|map| map.get(key).cloned())
2251 }
2252
2253 pub fn to_hash_map(&self) -> HashMap<K, V> {
2255 self.state.with(Clone::clone)
2256 }
2257
2258 pub fn insert(&self, key: K, value: V) -> Option<V> {
2260 self.state.update(|map| map.insert(key, value))
2261 }
2262
2263 pub fn extend<I>(&self, iter: I)
2265 where
2266 I: IntoIterator<Item = (K, V)>,
2267 {
2268 self.state.update(|map| map.extend(iter));
2269 }
2270
2271 pub fn remove(&self, key: &K) -> Option<V> {
2273 self.state.update(|map| map.remove(key))
2274 }
2275
2276 pub fn clear(&self) {
2278 self.state.replace(HashMap::default());
2279 }
2280
2281 pub fn retain<F>(&self, mut predicate: F)
2284 where
2285 F: FnMut(&K, &mut V) -> bool,
2286 {
2287 self.state.update(|map| map.retain(|k, v| predicate(k, v)));
2288 }
2289}
2290
2291impl<K, V> fmt::Debug for SnapshotStateMap<K, V>
2292where
2293 K: Clone + Eq + Hash + fmt::Debug + 'static,
2294 V: Clone + fmt::Debug + 'static,
2295{
2296 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2297 let contents = self.to_hash_map();
2298 f.debug_struct("SnapshotStateMap")
2299 .field("entries", &contents)
2300 .finish()
2301 }
2302}
2303
2304pub(crate) struct DerivedState<T: Clone + PartialEq + 'static> {
2307 core: Rc<DerivedCore<T>>,
2308}
2309
2310struct DerivedCore<T: Clone + PartialEq + 'static> {
2311 compute: RefCell<Rc<dyn Fn() -> T>>,
2312 state: OwnedMutableState<T>,
2313 derivation: Weak<RecomposeScopeInner>,
2316 refreshing: Cell<bool>,
2319}
2320
2321impl<T: Clone + PartialEq + 'static> DerivedState<T> {
2322 pub(crate) fn new(
2326 runtime: RuntimeHandle,
2327 compute: Rc<dyn Fn() -> T>,
2328 derivation: Option<&RecomposeScope>,
2329 ) -> Self {
2330 let initial = compute();
2331 let core = Rc::new(DerivedCore {
2332 compute: RefCell::new(compute),
2333 state: OwnedMutableState::with_runtime_structural_eq(initial, runtime),
2334 derivation: derivation.map_or_else(Weak::new, RecomposeScope::downgrade),
2335 refreshing: Cell::new(false),
2336 });
2337 let cell = core.state.handle();
2338 let refresh: Weak<dyn runtime::DerivedRefresh> = Rc::downgrade(&core) as Weak<_>;
2339 runtime::with_state_arena_by_id(cell.runtime_id, |arena| {
2340 arena.register_derived(cell.id, refresh);
2341 });
2342 Self { core }
2343 }
2344
2345 pub(crate) fn set_compute(&self, compute: Rc<dyn Fn() -> T>) {
2346 *self.core.compute.borrow_mut() = compute;
2347 }
2348
2349 pub(crate) fn recompute(&self) {
2350 let value = self.core.compute()();
2351 self.core.state.set_value(value);
2352 }
2353
2354 pub(crate) fn state(&self) -> State<T> {
2356 self.core.state.as_state().of_derived()
2357 }
2358}
2359
2360impl<T: Clone + PartialEq + 'static> DerivedCore<T> {
2361 fn compute(&self) -> Rc<dyn Fn() -> T> {
2362 Rc::clone(&self.compute.borrow())
2363 }
2364}
2365
2366impl<T: Clone + PartialEq + 'static> runtime::DerivedRefresh for DerivedCore<T> {
2367 fn refresh(&self) {
2368 let stale =
2369 RecomposeScope::upgrade(&self.derivation).is_some_and(|scope| scope.is_invalid());
2370 if !stale || self.refreshing.replace(true) {
2371 return;
2372 }
2373 let value = crate::composer_context::without_composer(|| self.compute()());
2376 self.refreshing.set(false);
2377 self.state.set_value(value);
2378 }
2379}
2380
2381impl<T: fmt::Debug + Clone + 'static> fmt::Debug for State<T> {
2382 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2383 if let Some(value) = self.try_value() {
2384 f.debug_struct("State").field("value", &value).finish()
2385 } else {
2386 f.write_str("State { value: <unavailable> }")
2387 }
2388 }
2389}
2390
2391#[cfg(test)]
2392#[path = "tests/state_tests.rs"]
2393mod tests;