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::manager::{Lifecycle, encoded_size};
25use crate::storage::{ReadLimits, RecoveryPosition, StorageManager};
26use crate::{Error, Result};
27use harmos_domain::state::Position;
28
29/* === Constants === */
30
31/* === Macros === */
32
33/* === Type Aliases === */
34
35/* === Structs === */
36
37pub struct RecoveryReader<'a, C> {
38    context: crate::execution::Context,
39    storage: &'a StorageManager<C>,
40    _lifecycle: RwLockReadGuard<'a, Lifecycle>,
41    through: RecoveryPosition,
42    limits: ReadLimits,
43    snapshot_bound: Option<NonZeroUsize>,
44}
45
46/* === Enums === */
47
48/* === Traits === */
49
50/* === Implementations === */
51
52impl<C> StorageManager<C> {
53    #[tracing::instrument(name = "storage.state.reader.recovery_reader", level = "debug", skip_all)]
54    pub async fn recovery_reader(&self, limits: ReadLimits, snapshot_bytes: Option<NonZeroUsize>) -> Result<RecoveryReader<'_, C>> {
55        let lifecycle = tracing::Instrument::instrument(self.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait")).await;
56        let invalid = |reason| Error::InvalidInput { field: "state recovery", reason };
57        let snapshot_bound = snapshot_bytes;
58        let through = {
59            let mut current = self.inner.state.lock().unwrap();
60            current.collect();
61            if current.status.running || current.status.closing || current.journal_saved.is_some() {
62                return Err(invalid("managed recovery requires a dormant acquired manager"));
63            }
64            current.status.recovery_saved.clone()
65        };
66        Ok(RecoveryReader {
67            context: crate::execution::Context::current(),
68            storage: self,
69            _lifecycle: lifecycle,
70            through,
71            limits,
72            snapshot_bound,
73        })
74    }
75}
76
77impl<C> RecoveryReader<'_, C> {
78    pub fn through(&self) -> &RecoveryPosition {
79        &self.through
80    }
81
82    pub async fn snapshot(&self) -> Result<Option<crate::ReadBatch<RecoverySnapshot>>> {
83        let context = self.context.clone();
84        let span = context.sync(|| tracing::trace_span!("storage.reader.page"));
85        let future = async {
86            let Some(bound) = self.snapshot_bound else {
87                return Ok(None);
88            };
89            let snapshot = self.storage.inner.bounded("recovery.read_snapshot", self.storage.inner.recovery.read_snapshot(bound)).await?;
90            if let Some(snapshot) = &snapshot
91                && (!self.through.covers(&snapshot.through)? || encoded_size(snapshot)? > bound.get())
92            {
93                return Err(Error::InvalidInput {
94                    field: "state snapshot",
95                    reason: "invalid managed snapshot boundary or size",
96                });
97            }
98            if let Some(snapshot) = &snapshot {
99                self.storage.check_state_dependency(&snapshot.required_data)?;
100            }
101            Ok(snapshot.map(crate::ReadBatch::new))
102        };
103        context.scope(tracing::Instrument::instrument(future, span)).await
104    }
105
106    pub async fn page(&self, after: &RecoveryPosition) -> Result<crate::ReadBatch<RecoveryPage>> {
107        let context = self.context.clone();
108        let span = context.sync(|| tracing::trace_span!("storage.reader.page"));
109        let future = async {
110            let page = self
111                .storage
112                .inner
113                .bounded("recovery.read_entries", self.storage.inner.recovery.read_entries(after, &self.through, self.limits))
114                .await?;
115            page.check(after, &self.through, self.limits, |entry| encoded_size(entry).unwrap_or(usize::MAX))?;
116            for entry in &page.entries {
117                self.storage.check_state_dependency(&entry.required_data)?;
118            }
119            Ok(crate::ReadBatch::new(page))
120        };
121        context.scope(tracing::Instrument::instrument(future, span)).await
122    }
123
124    pub fn finish(self, position: Position) -> watch::Receiver<Position> {
125        let (saved, durable) = watch::channel(position);
126        let mut current = self.storage.inner.state.lock().unwrap();
127        current.journal_required_data = current.status.saved.clone();
128        current.journal_saved = Some(saved);
129        current.journal_limit = self.limits.bytes.get();
130        durable
131    }
132}
133
134/* === Public Functions === */
135
136/* === Private Functions === */
137
138/* === Tests === */
harmos_storage/state/reader.rs

harmos_storage/state/
reader.rs
