cranpose_services/
durable_save.rs

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

cranpose_services/
durable_save.rs
