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

harmos_runtime/state/
snapshot.rs
