harmos_storage/state/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod admission;
13pub mod config;
14pub mod file;
15mod reader;
16
17/* === Re-Exports === */
18
19pub use admission::Permit;
20pub use config::StorageStateSettings;
21pub use reader::RecoveryReader;
22
23/* === Imports === */
24
25// Standard Library Imports
26
27// Third Party Imports
28use serde::{Deserialize, Serialize};
29// Local Imports
30use super::{ReadLimits, RecoveryPosition, StorageFuture, StoreHistory};
31use crate::{Error, Result};
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39/* === Structs === */
40
41#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
42#[serde(deny_unknown_fields)]
43pub struct RecoveryEntry {
44    pub required_data: super::StorageCursor,
45    pub position: RecoveryPosition,
46
47    #[serde(with = "payload")]
48    pub payload: Vec<u8>,
49}
50
51#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
52#[serde(deny_unknown_fields)]
53pub struct RecoverySnapshot {
54    pub required_data: super::StorageCursor,
55    pub through: RecoveryPosition,
56
57    #[serde(with = "payload")]
58    pub payload: Vec<u8>,
59}
60
61#[derive(Clone, Debug, PartialEq, Eq)]
62pub struct RecoveryPage {
63    pub entries: Vec<RecoveryEntry>,
64    pub next: Option<RecoveryPosition>,
65}
66
67/* === Enums === */
68
69/* === Traits === */
70
71pub trait StateBackend: Send + Sync + 'static {
72    fn history(&self) -> &StoreHistory;
73    fn open(&self) -> StorageFuture<'_, RecoveryPosition>;
74    fn read_snapshot(&self, max_bytes: std::num::NonZeroUsize) -> StorageFuture<'_, Option<RecoverySnapshot>>;
75    fn write_snapshot<'a>(&'a self, snapshot: &'a RecoverySnapshot) -> StorageFuture<'a, ()>;
76    fn read_entries<'a>(
77        &'a self, after: &'a RecoveryPosition, through: &'a RecoveryPosition, limits: ReadLimits,
78    ) -> StorageFuture<'a, RecoveryPage>;
79    fn write_entries<'a>(&'a self, entries: &'a [RecoveryEntry]) -> StorageFuture<'a, ()>;
80    fn flush<'a>(&'a self, through: &'a RecoveryPosition) -> StorageFuture<'a, RecoveryPosition>;
81    fn close(&self) -> StorageFuture<'_, ()>;
82}
83
84/* === Implementations === */
85
86impl RecoveryPage {
87    pub fn check(
88        &self, after: &RecoveryPosition, through: &RecoveryPosition, limits: ReadLimits,
89        mut encoded_bytes: impl FnMut(&RecoveryEntry) -> usize,
90    ) -> Result<()> {
91        if !through.covers(after)? {
92            return Err(Error::InvalidInput { field: "recovery range", reason: "start exceeds fixed frontier" });
93        }
94        let mut previous = after.sequence;
95        let mut bytes = 0usize;
96
97        for entry in &self.entries {
98            after.history.check(&entry.position.history)?;
99            let expected = previous.checked_add(1).ok_or(Error::InvalidInput { field: "recovery page", reason: "position overflow" })?;
100            if entry.position.sequence < expected {
101                return Err(Error::InvalidInput { field: "recovery page", reason: "duplicate or reversed position" });
102            }
103            if entry.position.sequence > expected {
104                return Err(Error::Gap { requested: expected, available: entry.position.sequence });
105            }
106            if entry.position.sequence > through.sequence {
107                return Err(Error::InvalidInput { field: "recovery page", reason: "entry exceeds fixed frontier" });
108            }
109            bytes = bytes
110                .checked_add(encoded_bytes(entry))
111                .ok_or(Error::InvalidInput { field: "recovery page", reason: "encoded size overflow" })?;
112            previous = entry.position.sequence;
113        }
114        limits.check(self.entries.len(), bytes)?;
115
116        let expected_next =
117            if previous < through.sequence { Some(RecoveryPosition { history: after.history.clone(), sequence: previous }) } else { None };
118        if self.next != expected_next || (self.entries.is_empty() && expected_next.is_some()) {
119            return Err(Error::InvalidInput { field: "recovery page", reason: "missing progress or invalid continuation" });
120        }
121
122        Ok(())
123    }
124}
125
126/* === Public Functions === */
127
128/* === Private Functions === */
129
130// Opaque Payloads Encode As Unpadded Base64, So Every Byte Changes The Encoded Size.
131mod payload {
132    use base64::Engine;
133    use base64::engine::general_purpose::STANDARD_NO_PAD;
134    use serde::{Deserialize, Deserializer, Serializer, de::Error};
135
136    pub(super) fn serialize<S: Serializer>(payload: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
137        serializer.serialize_str(&STANDARD_NO_PAD.encode(payload))
138    }
139
140    pub(super) fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
141        let encoded = String::deserialize(deserializer)?;
142        STANDARD_NO_PAD.decode(encoded.as_bytes()).map_err(D::Error::custom)
143    }
144}
145
146/* === Tests === */
harmos_storage/state/mod.rs

harmos_storage/state/
mod.rs
