harmos_runtime/state/
snapshot.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::collections::{HashMap, VecDeque};
18use std::future::{Future, poll_fn};
19use std::num::{NonZeroU64, NonZeroUsize};
20use std::panic::{AssertUnwindSafe, catch_unwind};
21use std::pin::pin;
22use std::sync::{
23    Arc,
24    atomic::{AtomicU64, Ordering},
25};
26use std::task::Poll;
27// Third Party Imports
28use serde::{Serialize, de::DeserializeOwned};
29// Local Imports
30use super::snapshot_codec;
31use crate::state::{Tail, TrackedState, Undos, dispatch::Dispatch};
32use crate::storage::{SchemaKey, StorageManager};
33use crate::{
34    Application, Error, Result, runtime::Incompatible, state::At, state::Entry, state::Position, state::Receipt, state::Recorded,
35    state::RequestKey, storage::Payload,
36};
37
38/* === Constants === */
39
40/* === Macros === */
41
42/* === Type Aliases === */
43
44type Encoding<A> =
45    fn(&SnapshotPolicy, &At<<A as Application>::State>, &Undos<A>, &Tail<A>, &HashMap<RequestKey, Receipt>) -> Result<Vec<u8>>;
46type Decoding<A> = fn(&[u8], &SnapshotPolicy, &Dispatch<A, Payload<A>>, usize) -> std::result::Result<Restored<A>, Incompatible>;
47
48/* === Structs === */
49
50#[derive(Clone, Debug)]
51pub struct SnapshotPolicy {
52    pub schema: SchemaKey,
53    pub every: NonZeroU64,
54    pub max_bytes: NonZeroUsize,
55}
56
57pub(crate) struct Snapshots<A: Application> {
58    pub(crate) policy: SnapshotPolicy,
59    encode: Encoding<A>,
60    pub(crate) decode: Decoding<A>,
61    last: AtomicU64,
62}
63
64pub(crate) struct Restored<A: Application> {
65    pub(crate) state: At<A::State>,
66    pub(crate) undos: Undos<A>,
67    pub(crate) receipts: HashMap<RequestKey, Receipt>,
68    pub(crate) tail: VecDeque<Arc<Entry<Recorded<A>>>>,
69}
70
71/* === Enums === */
72
73/* === Traits === */
74
75/* === Implementations === */
76
77impl<A: Application> Snapshots<A> {
78    pub(crate) fn new(policy: SnapshotPolicy) -> Self
79    where
80        A::State: Serialize + DeserializeOwned,
81    {
82        Self { policy, encode: snapshot_codec::encode::<A>, decode: snapshot_codec::decode::<A>, last: AtomicU64::new(0) }
83    }
84
85    pub(crate) fn resumed(&self, position: Position) {
86        self.last.store(position.0, Ordering::Relaxed);
87    }
88
89    pub(crate) async fn capture(
90        &self, storage: &StorageManager<crate::storage::ErasedCatalog>, tracked: &TrackedState<A>, undos: &Undos<A>,
91        receipts: &HashMap<RequestKey, Receipt>, final_save: bool,
92    ) {
93        if tracked.poisoned() {
94            return;
95        }
96        let position = tracked.state.read().await.position;
97        let last = self.last.load(Ordering::Relaxed);
98        if position.0 <= last || (!final_save && position.0 - last < self.policy.every.get()) {
99            return;
100        }
101        // The journal writer is the only mutator. Reserve before holding readers;
102        // no snapshot worker, detached state copy, or second persistence queue.
103        let result = async {
104            if final_save {
105                // Shutdown cannot wait for the periodic saver to free queue space.
106                storage.flush().await?;
107                storage.flush_recovery().await?;
108            }
109            let mut sealing = tracked.sealing.subscribe();
110            let reservation = storage.reserve_snapshot(self.policy.max_bytes.get());
111            let mut permit =
112                if final_save {
113                    reservation.await?
114                } else {
115                    if *sealing.borrow() {
116                        return Ok(());
117                    }
118                    let mut reservation = pin!(reservation);
119                    let mut changed = pin!(sealing.changed());
120                    let Some(permit) = poll_fn(|cx| {
121                        if changed.as_mut().poll(cx).is_ready() { Poll::Ready(None) } else { reservation.as_mut().poll(cx).map(Some) }
122                    })
123                    .await
124                    else {
125                        return Ok(());
126                    };
127                    permit?
128                };
129            let state = tracked.state.read().await;
130            let tail = tracked.tail.read().await;
131            self.last.store(position.0, Ordering::Relaxed);
132            let payload = catch_unwind(AssertUnwindSafe(|| (self.encode)(&self.policy, &state, undos, &tail, receipts)))
133                .map_err(|_| Error::InvalidInput { field: "state snapshot", reason: "serializer panicked" })??;
134            permit.snapshot_payload(payload)?;
135            permit.commit(position);
136            Ok::<(), Error>(())
137        }
138        .await;
139        if let Err(error) = result {
140            let error = harmos_domain::failure::Report(&error);
141            tracing::warn!(%error, position = position.0, "snapshot skipped; recovery retains the journal");
142        }
143    }
144}
145
146/* === Public Functions === */
147
148/* === Private Functions === */
149
150/* === Tests === */
harmos_runtime/state/snapshot.rs

harmos_runtime/state/
snapshot.rs
