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

harmos_storage/state/
mod.rs
