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 {
83            policy,
84            encode: snapshot_codec::encode::<A>,
85            decode: snapshot_codec::decode::<A>,
86            last: AtomicU64::new(0),
87        }
88    }
89
90    pub(crate) fn resumed(&self, position: Position) {
91        self.last.store(position.0, Ordering::Relaxed);
92    }
93
94    #[tracing::instrument(target = "harmos::state", name = "state.capture", level = "debug", skip_all)]
95    pub(crate) async fn capture(
96        &self, storage: &StorageManager<crate::storage::ErasedCatalog>, tracked: &TrackedState<A>, undos: &Undos<A>,
97        receipts: &HashMap<RequestKey, Receipt>, final_save: bool,
98    ) {
99        if tracked.poisoned() {
100            return;
101        }
102        let position = tracked.state.read().await.position;
103        let last = self.last.load(Ordering::Relaxed);
104        if position.0 <= last || (!final_save && position.0 - last < self.policy.every.get()) {
105            return;
106        }
107        // The journal writer is the only mutator. Reserve before holding readers;
108        // no snapshot worker, detached state copy, or second persistence queue.
109        let result = async {
110            if final_save {
111                // Shutdown cannot wait for the periodic saver to free queue space.
112                storage.flush().await?;
113                storage.flush_recovery().await?;
114            }
115            let mut sealing = tracked.sealing.subscribe();
116            let reservation = storage.reserve_snapshot(self.policy.max_bytes.get());
117            let mut permit =
118                if final_save {
119                    reservation.await?
120                } else {
121                    if *sealing.borrow() {
122                        return Ok(());
123                    }
124                    let mut reservation = pin!(reservation);
125                    let mut changed = pin!(sealing.changed());
126                    let Some(permit) = poll_fn(|cx| {
127                        if changed.as_mut().poll(cx).is_ready() { Poll::Ready(None) } else { reservation.as_mut().poll(cx).map(Some) }
128                    })
129                    .await
130                    else {
131                        return Ok(());
132                    };
133                    permit?
134                };
135            let state = tracked.state.read().await;
136            let tail = tracked.tail.read().await;
137            self.last.store(position.0, Ordering::Relaxed);
138            let payload =
139                catch_unwind(AssertUnwindSafe(|| (self.encode)(&self.policy, &state, undos, &tail, receipts))).map_err(|_| {
140                    Error::InvalidInput {
141                        field: "state snapshot",
142                        reason: "serializer panicked",
143                    }
144                })??;
145            permit.snapshot_payload(payload)?;
146            permit.commit(position);
147            Ok::<(), Error>(())
148        }
149        .await;
150        if let Err(error) = result {
151            let error = harmos_domain::failure::Report(&error);
152            tracing::warn!(%error, position = position.0, "snapshot skipped; recovery retains the journal");
153        }
154    }
155}
156
157/* === Public Functions === */
158
159/* === Private Functions === */
160
161/* === Tests === */
harmos_runtime/state/snapshot.rs

harmos_runtime/state/
snapshot.rs
