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    #[tracing::instrument(name = "storage.manager.recovery.write_recovery", level = "debug", skip_all)]
39    pub async fn write_recovery(&self, payloads: Vec<Vec<u8>>) -> Result<RecoveryPosition> {
40        self.check_raw_recovery()?;
41        if payloads.is_empty() {
42            return Ok(self.status().recovery_accepted);
43        }
44        let required_data = self.cursor_saved();
45        let bytes = payloads.iter().try_fold(0usize, |bytes, payload| {
46            bytes.checked_add(recovery_encoded_bound(payload.len(), self.inner.recovery.history(), self.inner.primary.history())?).ok_or(
47                Error::InvalidInput {
48                    field: "storage admission",
49                    reason: "size overflow",
50                },
51            )
52        })?;
53        let sequence = self
54            .admit(Domain::Recovery, bytes, payloads.len(), |first| {
55                Ok(Operation::Recovery(crate::StorageBatch::new(
56                    payloads
57                        .into_iter()
58                        .enumerate()
59                        .map(|(index, payload)| RecoveryEntry {
60                            required_data: required_data.clone(),
61                            position: self.recovery_position(first + index as u64),
62                            payload,
63                        })
64                        .collect(),
65                )))
66            })
67            .await?;
68        Ok(self.recovery_position(sequence))
69    }
70
71    #[tracing::instrument(name = "storage.manager.recovery.write_snapshot", level = "debug", skip_all)]
72    pub async fn write_snapshot(&self, snapshot: RecoverySnapshot) -> Result<()> {
73        self.check_raw_recovery()?;
74        self.inner.recovery.history().check(&snapshot.through.history)?;
75        self.check_state_dependency(&snapshot.required_data)?;
76        let bytes = recovery_encoded_bound(snapshot.payload.len(), &snapshot.through.history, &snapshot.required_data.history)?;
77        self.admit(Domain::Recovery, bytes, 0, |_| Ok(Operation::Snapshot(StorageBatch::new(vec![snapshot])))).await?;
78        Ok(())
79    }
80
81    #[tracing::instrument(name = "storage.manager.recovery.flush", level = "debug", skip_all)]
82    pub async fn flush(&self) -> Result<StorageCursor> {
83        let through = self.cursor_accepted();
84        self.flush_until(&through).await
85    }
86
87    #[tracing::instrument(name = "storage.manager.recovery.flush_until", level = "debug", skip_all)]
88    pub async fn flush_until(&self, through: &StorageCursor) -> Result<StorageCursor> {
89        let files = self.capture_files(Some(through))?;
90        self.flush_prepared(through, &files).await
91    }
92
93    #[tracing::instrument(name = "storage.manager.recovery.flush_prepared", level = "debug", skip_all)]
94    pub async fn flush_prepared(&self, through: &StorageCursor, files: &Option<crate::FileSave>) -> Result<StorageCursor> {
95        self.flush_domain(Domain::Data, Some(through)).await?;
96        if files.is_some() {
97            let _lifecycle = tracing::Instrument::instrument(self.inner.lifecycle.write(), tracing::trace_span!("storage.lock.wait")).await;
98            if self.inner.state.lock().unwrap().status.closed {
99                return Err(Error::Closing { domain: "storage" });
100            }
101            self.persist_files(files).await?;
102        }
103        Ok(self.cursor_saved())
104    }
105
106    #[tracing::instrument(name = "storage.manager.recovery.flush_recovery", level = "debug", skip_all)]
107    pub async fn flush_recovery(&self) -> Result<RecoveryPosition> {
108        self.flush_domain(Domain::Recovery, None).await?;
109        Ok(self.status().recovery_saved)
110    }
111
112    #[tracing::instrument(name = "storage.manager.recovery.read_recovery", level = "debug", skip_all)]
113    pub async fn read_recovery(
114        &self, after: &RecoveryPosition, through: &RecoveryPosition, limits: ReadLimits,
115    ) -> Result<crate::ReadBatch<RecoveryPage>> {
116        let _lifecycle = tracing::Instrument::instrument(self.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait")).await;
117        let status = self.status();
118        if !status.recovery_saved.covers(through)? {
119            return Err(Error::InvalidInput {
120                field: "recovery range",
121                reason: "frontier is not durable",
122            });
123        }
124        let page = self.inner.bounded("recovery.read_entries", self.inner.recovery.read_entries(after, through, limits)).await?;
125        page.check(after, through, limits, |entry| encoded_size(entry).unwrap_or(usize::MAX))?;
126        for entry in &page.entries {
127            self.check_state_dependency(&entry.required_data)?;
128        }
129        Ok(crate::ReadBatch::new(page))
130    }
131
132    #[tracing::instrument(name = "storage.manager.recovery.read_snapshot", level = "debug", skip_all)]
133    pub async fn read_snapshot(&self, max_bytes: std::num::NonZeroUsize) -> Result<Option<crate::ReadBatch<RecoverySnapshot>>> {
134        let _lifecycle = tracing::Instrument::instrument(self.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait")).await;
135        let snapshot = self.inner.bounded("recovery.read_snapshot", self.inner.recovery.read_snapshot(max_bytes)).await?;
136        if let Some(snapshot) = &snapshot
137            && (!self.status().recovery_saved.covers(&snapshot.through)?
138                || !self.status().saved.covers(&snapshot.required_data)?
139                || encoded_size(snapshot)? > max_bytes.get())
140        {
141            return Err(Error::InvalidInput {
142                field: "recovery snapshot",
143                reason: "invalid saved boundary or byte size",
144            });
145        }
146        Ok(snapshot.map(crate::ReadBatch::new))
147    }
148}
149
150/* === Public Functions === */
151
152/* === Private Functions === */
153
154/* === Tests === */
harmos_storage/manager/recovery.rs

harmos_storage/manager/
recovery.rs
