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, StorageBatch, StorageFuture, StoreHistory};
31use crate::{Error, MemorySize, 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(&self, snapshot: StorageBatch<RecoverySnapshot>) -> StorageFuture<'_, ()>;
76    fn read_entries<'a>(
77        &'a self, after: &'a RecoveryPosition, through: &'a RecoveryPosition, limits: ReadLimits,
78    ) -> StorageFuture<'a, RecoveryPage>;
79    fn write_entries(&self, entries: StorageBatch<RecoveryEntry>) -> StorageFuture<'_, ()>;
80    fn flush<'a>(&'a self, through: &'a RecoveryPosition) -> StorageFuture<'a, RecoveryPosition>;
81    fn close(&self) -> StorageFuture<'_, ()>;
82}
83
84/* === Implementations === */
85
86impl MemorySize for RecoveryEntry {
87    fn memory_size(&self) -> usize {
88        std::mem::size_of::<Self>()
89            .saturating_add(self.required_data.memory_size().saturating_sub(std::mem::size_of::<super::StorageCursor>()))
90            .saturating_add(self.position.history.memory_size().saturating_sub(std::mem::size_of::<StoreHistory>()))
91            .saturating_add(self.payload.capacity())
92    }
93}
94
95impl MemorySize for RecoverySnapshot {
96    fn memory_size(&self) -> usize {
97        std::mem::size_of::<Self>()
98            .saturating_add(self.required_data.memory_size().saturating_sub(std::mem::size_of::<super::StorageCursor>()))
99            .saturating_add(self.through.history.memory_size().saturating_sub(std::mem::size_of::<StoreHistory>()))
100            .saturating_add(self.payload.capacity())
101    }
102}
103
104impl MemorySize for RecoveryPage {
105    fn memory_size(&self) -> usize {
106        std::mem::size_of::<Self>()
107            .saturating_add(self.entries.memory_size().saturating_sub(std::mem::size_of::<Vec<RecoveryEntry>>()))
108            .saturating_add(
109                self.next.as_ref().map_or(0, |next| next.history.memory_size().saturating_sub(std::mem::size_of::<StoreHistory>())),
110            )
111    }
112}
113
114impl RecoveryPage {
115    pub fn check(
116        &self, after: &RecoveryPosition, through: &RecoveryPosition, limits: ReadLimits,
117        mut encoded_bytes: impl FnMut(&RecoveryEntry) -> usize,
118    ) -> Result<()> {
119        if !through.covers(after)? {
120            return Err(Error::InvalidInput {
121                field: "recovery range",
122                reason: "start exceeds fixed frontier",
123            });
124        }
125        let mut previous = after.sequence;
126        let mut bytes = 0usize;
127
128        for entry in &self.entries {
129            after.history.check(&entry.position.history)?;
130            let expected = previous.checked_add(1).ok_or(Error::InvalidInput {
131                field: "recovery page",
132                reason: "position overflow",
133            })?;
134            if entry.position.sequence < expected {
135                return Err(Error::InvalidInput {
136                    field: "recovery page",
137                    reason: "duplicate or reversed position",
138                });
139            }
140            if entry.position.sequence > expected {
141                return Err(Error::Gap {
142                    requested: expected,
143                    available: entry.position.sequence,
144                });
145            }
146            if entry.position.sequence > through.sequence {
147                return Err(Error::InvalidInput {
148                    field: "recovery page",
149                    reason: "entry exceeds fixed frontier",
150                });
151            }
152            bytes = bytes.checked_add(encoded_bytes(entry)).ok_or(Error::InvalidInput {
153                field: "recovery page",
154                reason: "encoded size overflow",
155            })?;
156            previous = entry.position.sequence;
157        }
158        limits.check(self.entries.len(), bytes)?;
159
160        let expected_next = if previous < through.sequence {
161            Some(RecoveryPosition {
162                history: after.history.clone(),
163                sequence: previous,
164            })
165        } else {
166            None
167        };
168        if self.next != expected_next || (self.entries.is_empty() && expected_next.is_some()) {
169            return Err(Error::InvalidInput {
170                field: "recovery page",
171                reason: "missing progress or invalid continuation",
172            });
173        }
174
175        Ok(())
176    }
177}
178
179/* === Public Functions === */
180
181/* === Private Functions === */
182
183// Opaque Payloads Encode As Unpadded Base64, So Every Byte Changes The Encoded Size.
184mod payload {
185    use base64::Engine;
186    use base64::engine::general_purpose::STANDARD_NO_PAD;
187    use serde::{Deserialize, Deserializer, Serializer, de::Error};
188
189    pub(super) fn serialize<S: Serializer>(payload: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
190        serializer.serialize_str(&STANDARD_NO_PAD.encode(payload))
191    }
192
193    pub(super) fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
194        let encoded = String::deserialize(deserializer)?;
195        STANDARD_NO_PAD.decode(encoded.as_bytes()).map_err(D::Error::custom)
196    }
197}
198
199/* === Tests === */
harmos_storage/state/mod.rs

harmos_storage/state/
mod.rs
