harmos_storage/state/
reader.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
17use std::num::NonZeroUsize;
18
19// Third Party Imports
20use tokio::sync::{RwLockReadGuard, watch};
21
22// Local Imports
23use super::{RecoveryPage, RecoverySnapshot};
24use crate::storage::capacity::{Pool, Reservation};
25use crate::storage::manager::{ENTRY_OVERHEAD, Lifecycle, encoded_size, recovery_preparation_bytes};
26use crate::storage::{ReadLimits, RecoveryPosition, StorageManager};
27use crate::{Error, Result};
28use harmos_domain::state::Position;
29
30/* === Constants === */
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36/* === Structs === */
37
38pub struct RecoveryReader<'a, C> {
39    storage: &'a StorageManager<C>,
40    // Release Budget Before Unlocking The Recovery Lifecycle.
41    _reservation: Reservation,
42    _lifecycle: RwLockReadGuard<'a, Lifecycle>,
43    through: RecoveryPosition,
44    limits: ReadLimits,
45    snapshot_bound: Option<NonZeroUsize>,
46}
47
48/* === Enums === */
49
50/* === Traits === */
51
52/* === Implementations === */
53
54impl<C> StorageManager<C> {
55    pub async fn recovery_reader(&self, limits: ReadLimits, snapshot_bytes: Option<NonZeroUsize>) -> Result<RecoveryReader<'_, C>> {
56        let lifecycle = self.inner.lifecycle.read().await;
57        let invalid = |reason| Error::InvalidInput { field: "state recovery", reason };
58        let snapshot_bound = snapshot_bytes
59            .map(|bytes| {
60                recovery_preparation_bytes(bytes.get(), self.inner.recovery.history(), self.inner.primary.history())
61                    .map(|charged| NonZeroUsize::new(charged).expect("positive snapshot budget"))
62            })
63            .transpose()?;
64        let mut bytes = limits
65            .bytes
66            .get()
67            .checked_mul(2)
68            .and_then(|n| n.checked_add(limits.rows.get().checked_mul(ENTRY_OVERHEAD)?))
69            .ok_or_else(|| invalid("recovery read size overflow"))?;
70        if let Some(bound) = snapshot_bound {
71            bytes = bytes.max(bound.get().checked_mul(2).ok_or_else(|| invalid("snapshot read size overflow"))?);
72        }
73        let through = {
74            let mut current = self.inner.state.lock().unwrap();
75            if current.status.running || current.status.closing || current.journal_saved.is_some() {
76                return Err(invalid("managed recovery requires a dormant acquired manager"));
77            }
78            if bytes > self.inner.settings.request.max_pending_write_bytes.get().saturating_sub(current.status.pending_bytes) {
79                return Err(invalid("recovery page exceeds shared memory budget"));
80            }
81            current.scratch_bytes += bytes;
82            current.collect();
83            current.status.recovery_saved.clone()
84        };
85        Ok(RecoveryReader {
86            storage: self,
87            _lifecycle: lifecycle,
88            _reservation: Reservation::charged(&self.inner, Pool::Write, bytes),
89            through,
90            limits,
91            snapshot_bound,
92        })
93    }
94}
95
96impl<C> RecoveryReader<'_, C> {
97    pub fn through(&self) -> &RecoveryPosition {
98        &self.through
99    }
100
101    pub async fn snapshot(&self) -> Result<Option<RecoverySnapshot>> {
102        let Some(bound) = self.snapshot_bound else {
103            return Ok(None);
104        };
105        let snapshot = self.storage.inner.bounded(self.storage.inner.recovery.read_snapshot(bound)).await?;
106        if let Some(snapshot) = &snapshot
107            && (!self.through.covers(&snapshot.through)? || encoded_size(snapshot)? > bound.get())
108        {
109            return Err(Error::InvalidInput { field: "state snapshot", reason: "invalid managed snapshot boundary or size" });
110        }
111        if let Some(snapshot) = &snapshot {
112            self.storage.check_state_dependency(&snapshot.required_data)?;
113        }
114        Ok(snapshot)
115    }
116
117    pub async fn page(&self, after: &RecoveryPosition) -> Result<RecoveryPage> {
118        let page = self.storage.inner.bounded(self.storage.inner.recovery.read_entries(after, &self.through, self.limits)).await?;
119        page.check(after, &self.through, self.limits, |entry| encoded_size(entry).unwrap_or(usize::MAX))?;
120        for entry in &page.entries {
121            self.storage.check_state_dependency(&entry.required_data)?;
122        }
123        Ok(page)
124    }
125
126    pub fn finish(self, position: Position) -> watch::Receiver<Position> {
127        let (saved, durable) = watch::channel(position);
128        let mut current = self.storage.inner.state.lock().unwrap();
129        current.journal_required_data = current.status.saved.clone();
130        current.journal_saved = Some(saved);
131        current.journal_limit = self.limits.bytes.get();
132        durable
133    }
134}
135
136/* === Public Functions === */
137
138/* === Private Functions === */
139
140/* === Tests === */
harmos_storage/state/reader.rs

harmos_storage/state/
reader.rs
