cranpose_services/
durable_save.rs

1use std::{
2    collections::BTreeMap,
3    panic::{AssertUnwindSafe, catch_unwind},
4    sync::{Arc, Mutex, OnceLock, PoisonError, atomic::Ordering},
5    time::Duration,
6};
7
8use portable_atomic::AtomicU64;
9
10use crate::preferences::{PreferencesError, PreferencesRef};
11
12/// Result of a lifecycle or explicit save boundary.
13#[derive(Clone, Copy, Debug, Eq, PartialEq)]
14pub enum DurableSaveOutcome {
15    /// No work was registered.
16    Nothing,
17    /// Every callback and preferences batch succeeded.
18    Completed,
19    /// A callback panicked, a worker could not start, or a write failed.
20    Failed,
21    /// The deadline expired. Running work can finish under its background lease.
22    TimedOut,
23}
24
25/// An error reported by a durable-save callback.
26#[derive(Clone, Debug, thiserror::Error)]
27pub enum DurableSaveError {
28    /// A preferences backend rejected a write.
29    #[error(transparent)]
30    Preferences(#[from] PreferencesError),
31    /// Application-specific durable storage failed.
32    #[error("durable save failed: {0}")]
33    Storage(String),
34}
35
36type SaveCallback = dyn Fn() -> Result<(), DurableSaveError> + Send + Sync;
37type PreferenceCallback = dyn Fn() -> String + Send + Sync;
38
39enum SaveWork {
40    Callback(Box<SaveCallback>),
41    Preference {
42        store: PreferencesRef,
43        key: String,
44        read: Box<PreferenceCallback>,
45    },
46}
47
48static NEXT_SAVE: AtomicU64 = AtomicU64::new(1);
49type Saves = BTreeMap<u64, Arc<SaveWork>>;
50
51fn saves() -> &'static Mutex<Saves> {
52    static SAVES: OnceLock<Mutex<Saves>> = OnceLock::new();
53    SAVES.get_or_init(Mutex::default)
54}
55
56/// Keeps a shared durable save registered until its owner is dropped.
57pub struct DurableSaveRegistration {
58    id: u64,
59}
60
61impl Drop for DurableSaveRegistration {
62    fn drop(&mut self) {
63        let removed = saves()
64            .lock()
65            .unwrap_or_else(PoisonError::into_inner)
66            .remove(&self.id);
67        drop(removed);
68    }
69}
70
71fn register(work: SaveWork) -> DurableSaveRegistration {
72    let id = NEXT_SAVE.fetch_add(1, Ordering::Relaxed);
73    saves()
74        .lock()
75        .unwrap_or_else(PoisonError::into_inner)
76        .insert(id, Arc::new(work));
77    DurableSaveRegistration { id }
78}
79
80/// Stops the shared saves of `store` keys that `keys` forgets, such as a
81/// popped back-stack entry's saved state that a view model still holds.
82fn stop_preference_saves(store: &PreferencesRef, keys: &local_saves::Forget) {
83    let stopped: Vec<_> = saves()
84        .lock()
85        .unwrap_or_else(PoisonError::into_inner)
86        .extract_if(.., |_, work| match &**work {
87            SaveWork::Preference {
88                store: saved, key, ..
89            } => Arc::ptr_eq(saved, store) && keys.matches(key),
90            SaveWork::Callback(_) => false,
91        })
92        .collect();
93    drop(stopped);
94}
95
96/// Registers fallible work for the next save boundary. Callbacks run serially;
97/// timed-out runs are allowed to finish before another run can start.
98pub fn register_durable_save(
99    save: impl Fn() -> Result<(), DurableSaveError> + Send + Sync + 'static,
100) -> DurableSaveRegistration {
101    register(SaveWork::Callback(Box::new(save)))
102}
103
104/// Registers a value to be sampled when saving starts and batched with other
105/// keys using the same store. The callback must not perform storage I/O.
106pub fn register_preference_save(
107    store: PreferencesRef,
108    key: String,
109    read: impl Fn() -> String + Send + Sync + 'static,
110) -> DurableSaveRegistration {
111    register(SaveWork::Preference {
112        store,
113        key,
114        read: Box::new(read),
115    })
116}
117
118/// Registers fallible durable work for this composition's lifetime.
119#[expect(non_snake_case)]
120#[track_caller]
121pub fn DurableSaveEffect<K: PartialEq + 'static>(
122    keys: K,
123    save: impl Fn() -> Result<(), DurableSaveError> + Send + Sync + 'static,
124) {
125    cranpose_core::__disposable_effect_impl(
126        cranpose_core::caller_location_key() ^ cranpose_core::location_key!(),
127        keys,
128        move |scope| {
129            let registration = register_durable_save(save);
130            scope.on_dispose(move || drop(registration))
131        },
132    );
133}
134
135mod local_saves;
136
137pub use local_saves::SaveableRegistration;
138pub(crate) use local_saves::{LocalSaves, forget, register as register_local, saved};
139
140struct PreferenceBatch {
141    store: PreferencesRef,
142    values: BTreeMap<String, String>,
143    forget: Vec<Arc<local_saves::Forget>>,
144}
145
146impl PreferenceBatch {
147    fn removed_keys(&self) -> Vec<String> {
148        if self.forget.is_empty() {
149            return Vec::new();
150        }
151        let mut keys = self.store.keys();
152        keys.retain(|key| self.forget.iter().any(|forget| forget.matches(key)));
153        keys
154    }
155}
156
157#[derive(Default)]
158struct SaveBatch {
159    preferences: Vec<PreferenceBatch>,
160    callbacks: Vec<Arc<SaveWork>>,
161}
162
163impl SaveBatch {
164    fn for_store(&mut self, store: &PreferencesRef) -> &mut PreferenceBatch {
165        let index = match self
166            .preferences
167            .iter()
168            .position(|batch| Arc::ptr_eq(&batch.store, store))
169        {
170            Some(index) => index,
171            None => {
172                self.preferences.push(PreferenceBatch {
173                    store: Arc::clone(store),
174                    values: BTreeMap::new(),
175                    forget: Vec::new(),
176                });
177                self.preferences.len() - 1
178            }
179        };
180        &mut self.preferences[index]
181    }
182
183    fn push(&mut self, store: &PreferencesRef, key: String, value: String) {
184        self.for_store(store).values.insert(key, value);
185    }
186
187    fn forget(&mut self, store: &PreferencesRef, keys: Arc<local_saves::Forget>) {
188        self.for_store(store).forget.push(keys);
189    }
190
191    /// The work to do now, and what of this thread's saves it covers.
192    fn collect() -> (Self, local_saves::Sampled) {
193        let mut batch = Self {
194            callbacks: saves()
195                .lock()
196                .unwrap_or_else(PoisonError::into_inner)
197                .values()
198                .cloned()
199                .collect(),
200            ..Self::default()
201        };
202        let sampled = local_saves::sample(&mut batch);
203        (batch, sampled)
204    }
205
206    fn is_empty(&self) -> bool {
207        self.preferences.is_empty() && self.callbacks.is_empty()
208    }
209
210    fn run(mut self) -> DurableSaveOutcome {
211        let mut succeeded = true;
212        for work in std::mem::take(&mut self.callbacks) {
213            succeeded &= match &*work {
214                SaveWork::Callback(save) => run_callback(save),
215                SaveWork::Preference { store, key, read } => {
216                    match catch_unwind(AssertUnwindSafe(read)) {
217                        Ok(value) => {
218                            self.push(store, key.clone(), value);
219                            true
220                        }
221                        Err(_) => false,
222                    }
223                }
224            };
225        }
226        for batch in self.preferences {
227            succeeded &= run_callback(|| {
228                batch
229                    .store
230                    .update_many(&batch.values, &batch.removed_keys())?;
231                batch.store.flush()?;
232                Ok(())
233            });
234        }
235        if succeeded {
236            DurableSaveOutcome::Completed
237        } else {
238            DurableSaveOutcome::Failed
239        }
240    }
241}
242
243fn run_callback(save: impl FnOnce() -> Result<(), DurableSaveError>) -> bool {
244    match catch_unwind(AssertUnwindSafe(save)) {
245        Ok(Ok(())) => true,
246        Ok(Err(error)) => {
247            log::warn!("cranpose: {error}");
248            false
249        }
250        Err(_) => {
251            log::warn!("cranpose: durable save panicked");
252            false
253        }
254    }
255}
256
257/// Captures composition-local values on the calling thread and saves registered
258/// values in one serial batch. Call this on each composition's owning thread.
259///
260/// Native runs use at most one worker with no queued batches. A concurrent call
261/// waits within its budget before sampling fresh values. A timeout does not
262/// cancel a running write. Retry a failed or timed-out boundary before relying
263/// on restoration. Web writes execute synchronously on the browser thread.
264#[inline(never)]
265pub fn run_durable_saves(deadline: Duration) -> DurableSaveOutcome {
266    #[cfg(not(target_arch = "wasm32"))]
267    {
268        executor::run(deadline)
269    }
270    #[cfg(target_arch = "wasm32")]
271    {
272        let started = web_time::Instant::now();
273        let (batch, sampled) = SaveBatch::collect();
274        if batch.is_empty() {
275            return DurableSaveOutcome::Nothing;
276        }
277        let result = batch.run();
278        if result == DurableSaveOutcome::Completed {
279            sampled.completed();
280        }
281        if result == DurableSaveOutcome::Completed && started.elapsed() > deadline {
282            DurableSaveOutcome::TimedOut
283        } else {
284            result
285        }
286    }
287}
288
289#[cfg(not(target_arch = "wasm32"))]
290mod executor;
cranpose_services/durable_save.rs

cranpose_services/
durable_save.rs
