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()
127            ^ cranpose_core::location_key(file!(), line!(), column!()),
128        keys,
129        move |scope| {
130            let registration = register_durable_save(save);
131            scope.on_dispose(move || drop(registration))
132        },
133    );
134}
135
136mod local_saves;
137
138pub use local_saves::SaveableRegistration;
139pub(crate) use local_saves::{LocalSaves, forget, register as register_local, saved};
140
141struct PreferenceBatch {
142    store: PreferencesRef,
143    values: BTreeMap<String, String>,
144    forget: Vec<Arc<local_saves::Forget>>,
145}
146
147impl PreferenceBatch {
148    fn removed_keys(&self) -> Vec<String> {
149        if self.forget.is_empty() {
150            return Vec::new();
151        }
152        let mut keys = self.store.keys();
153        keys.retain(|key| self.forget.iter().any(|forget| forget.matches(key)));
154        keys
155    }
156}
157
158#[derive(Default)]
159struct SaveBatch {
160    preferences: Vec<PreferenceBatch>,
161    callbacks: Vec<Arc<SaveWork>>,
162}
163
164impl SaveBatch {
165    fn for_store(&mut self, store: &PreferencesRef) -> &mut PreferenceBatch {
166        let index = match self
167            .preferences
168            .iter()
169            .position(|batch| Arc::ptr_eq(&batch.store, store))
170        {
171            Some(index) => index,
172            None => {
173                self.preferences.push(PreferenceBatch {
174                    store: Arc::clone(store),
175                    values: BTreeMap::new(),
176                    forget: Vec::new(),
177                });
178                self.preferences.len() - 1
179            }
180        };
181        &mut self.preferences[index]
182    }
183
184    fn push(&mut self, store: &PreferencesRef, key: String, value: String) {
185        self.for_store(store).values.insert(key, value);
186    }
187
188    fn forget(&mut self, store: &PreferencesRef, keys: Arc<local_saves::Forget>) {
189        self.for_store(store).forget.push(keys);
190    }
191
192    /// The work to do now, and what of this thread's saves it covers.
193    fn collect() -> (Self, local_saves::Sampled) {
194        let mut batch = Self {
195            callbacks: saves()
196                .lock()
197                .unwrap_or_else(PoisonError::into_inner)
198                .values()
199                .cloned()
200                .collect(),
201            ..Self::default()
202        };
203        let sampled = local_saves::sample(&mut batch);
204        (batch, sampled)
205    }
206
207    fn is_empty(&self) -> bool {
208        self.preferences.is_empty() && self.callbacks.is_empty()
209    }
210
211    fn run(mut self) -> DurableSaveOutcome {
212        let mut succeeded = true;
213        for work in std::mem::take(&mut self.callbacks) {
214            succeeded &= match &*work {
215                SaveWork::Callback(save) => run_callback(save),
216                SaveWork::Preference { store, key, read } => {
217                    match catch_unwind(AssertUnwindSafe(read)) {
218                        Ok(value) => {
219                            self.push(store, key.clone(), value);
220                            true
221                        }
222                        Err(_) => false,
223                    }
224                }
225            };
226        }
227        for batch in self.preferences {
228            succeeded &= run_callback(|| {
229                batch
230                    .store
231                    .update_many(&batch.values, &batch.removed_keys())?;
232                batch.store.flush()?;
233                Ok(())
234            });
235        }
236        if succeeded {
237            DurableSaveOutcome::Completed
238        } else {
239            DurableSaveOutcome::Failed
240        }
241    }
242}
243
244fn run_callback(save: impl FnOnce() -> Result<(), DurableSaveError>) -> bool {
245    match catch_unwind(AssertUnwindSafe(save)) {
246        Ok(Ok(())) => true,
247        Ok(Err(error)) => {
248            log::warn!("cranpose: {error}");
249            false
250        }
251        Err(_) => {
252            log::warn!("cranpose: durable save panicked");
253            false
254        }
255    }
256}
257
258/// Captures composition-local values on the calling thread and saves registered
259/// values in one serial batch. Call this on each composition's owning thread.
260///
261/// Native runs use at most one worker with no queued batches. A concurrent call
262/// waits within its budget before sampling fresh values. A timeout does not
263/// cancel a running write. Retry a failed or timed-out boundary before relying
264/// on restoration. Web writes execute synchronously on the browser thread.
265#[inline(never)]
266pub fn run_durable_saves(deadline: Duration) -> DurableSaveOutcome {
267    #[cfg(not(target_arch = "wasm32"))]
268    {
269        executor::run(deadline)
270    }
271    #[cfg(target_arch = "wasm32")]
272    {
273        let started = web_time::Instant::now();
274        let (batch, sampled) = SaveBatch::collect();
275        if batch.is_empty() {
276            return DurableSaveOutcome::Nothing;
277        }
278        let result = batch.run();
279        if result == DurableSaveOutcome::Completed {
280            sampled.completed();
281        }
282        if result == DurableSaveOutcome::Completed && started.elapsed() > deadline {
283            DurableSaveOutcome::TimedOut
284        } else {
285            result
286        }
287    }
288}
289
290#[cfg(not(target_arch = "wasm32"))]
291mod executor;
cranpose_services/durable_save.rs

cranpose_services/
durable_save.rs
