harmos_storage/manager/
recovery.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
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl<C> StorageManager<C> {
38    pub async fn write_recovery(&self, payloads: &[&[u8]]) -> Result<RecoveryPosition> {
39        self.check_raw_recovery()?;
40        if payloads.is_empty() {
41            return Ok(self.status().recovery_accepted);
42        }
43        let required_data = self.cursor_saved();
44        let bytes = payloads.iter().try_fold(0usize, |bytes, payload| {
45            bytes
46                .checked_add(recovery_resident_bytes(payload.len(), self.inner.recovery.history(), self.inner.primary.history())?)
47                .ok_or(Error::InvalidInput { field: "storage admission", reason: "size overflow" })
48        })?;
49        let sequence = self
50            .admit(Domain::Recovery, bytes, payloads.len(), |first| {
51                Ok(Operation::Recovery(
52                    payloads
53                        .iter()
54                        .enumerate()
55                        .map(|(index, payload)| RecoveryEntry {
56                            required_data: required_data.clone(),
57                            position: self.recovery_position(first + index as u64),
58                            payload: payload.to_vec(),
59                        })
60                        .collect(),
61                ))
62            })
63            .await?;
64        Ok(self.recovery_position(sequence))
65    }
66
67    pub async fn write_snapshot(&self, snapshot: &RecoverySnapshot) -> Result<()> {
68        self.check_raw_recovery()?;
69        self.inner.recovery.history().check(&snapshot.through.history)?;
70        self.check_state_dependency(&snapshot.required_data)?;
71        let bytes = recovery_resident_bytes(snapshot.payload.len(), &snapshot.through.history, &snapshot.required_data.history)?;
72        self.admit(Domain::Recovery, bytes, 0, |_| Ok(Operation::Snapshot(snapshot.clone()))).await?;
73        Ok(())
74    }
75
76    pub async fn flush(&self) -> Result<StorageCursor> {
77        let through = self.cursor_accepted();
78        self.flush_until(&through).await
79    }
80
81    pub async fn flush_until(&self, through: &StorageCursor) -> Result<StorageCursor> {
82        let files = self.capture_files(Some(through))?;
83        self.flush_prepared(through, &files).await
84    }
85
86    pub async fn flush_prepared(&self, through: &StorageCursor, files: &Option<crate::FileSave>) -> Result<StorageCursor> {
87        self.flush_domain(Domain::Data, Some(through)).await?;
88        if files.is_some() {
89            let _lifecycle = self.inner.lifecycle.write().await;
90            if self.inner.state.lock().unwrap().status.closed {
91                return Err(Error::Closing { domain: "storage" });
92            }
93            self.persist_files(files).await?;
94        }
95        Ok(self.cursor_saved())
96    }
97
98    pub async fn flush_recovery(&self) -> Result<RecoveryPosition> {
99        self.flush_domain(Domain::Recovery, None).await?;
100        Ok(self.status().recovery_saved)
101    }
102
103    pub async fn read_recovery(&self, after: &RecoveryPosition, through: &RecoveryPosition, limits: ReadLimits) -> Result<RecoveryPage> {
104        let _lifecycle = self.inner.lifecycle.read().await;
105        let charge = limits.rows.get().saturating_mul(ENTRY_OVERHEAD).saturating_add(limits.bytes.get());
106        let _reservation = Reservation::wait(&self.inner, Pool::Read, &Held::default(), charge, None).await.map_err(Refusal::into_error)?;
107        let status = self.status();
108        if !status.recovery_saved.covers(through)? {
109            return Err(Error::InvalidInput { field: "recovery range", reason: "frontier is not durable" });
110        }
111        let page = self.inner.bounded(self.inner.recovery.read_entries(after, through, limits)).await?;
112        page.check(after, through, limits, |entry| encoded_size(entry).unwrap_or(usize::MAX))?;
113        for entry in &page.entries {
114            self.check_state_dependency(&entry.required_data)?;
115        }
116        Ok(page)
117    }
118
119    pub async fn read_snapshot(&self, max_bytes: std::num::NonZeroUsize) -> Result<Option<RecoverySnapshot>> {
120        let _lifecycle = self.inner.lifecycle.read().await;
121        let charge = max_bytes.get().saturating_add(ENTRY_OVERHEAD);
122        let _reservation = Reservation::wait(&self.inner, Pool::Read, &Held::default(), charge, None).await.map_err(Refusal::into_error)?;
123        let snapshot = self.inner.bounded(self.inner.recovery.read_snapshot(max_bytes)).await?;
124        if let Some(snapshot) = &snapshot
125            && (!self.status().recovery_saved.covers(&snapshot.through)?
126                || !self.status().saved.covers(&snapshot.required_data)?
127                || encoded_size(snapshot)? > max_bytes.get())
128        {
129            return Err(Error::InvalidInput { field: "recovery snapshot", reason: "invalid saved boundary or byte size" });
130        }
131        Ok(snapshot)
132    }
133}
134
135/* === Public Functions === */
136
137/* === Private Functions === */
138
139/* === Tests === */
harmos_storage/manager/recovery.rs

harmos_storage/manager/
recovery.rs
