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

harmos_storage/manager/
recovery.rs
