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, recovery_encoded_bound};
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_bytes: 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 through = {
58            let mut current = self.inner.state.lock().unwrap();
59            current.collect();
60            if current.status.running || current.status.closing || current.journal_saved.is_some() {
61                return Err(invalid("managed recovery requires a dormant acquired manager"));
62            }
63            current.status.recovery_saved.clone()
64        };
65        Ok(RecoveryReader {
66            context: crate::execution::Context::current(),
67            storage: self,
68            _lifecycle: lifecycle,
69            through,
70            limits,
71            snapshot_bytes,
72        })
73    }
74}
75
76impl<C> RecoveryReader<'_, C> {
77    pub fn through(&self) -> &RecoveryPosition {
78        &self.through
79    }
80
81    pub async fn snapshot(&self) -> Result<Option<crate::ReadBatch<RecoverySnapshot>>> {
82        let context = self.context.clone();
83        let span = context.sync(|| tracing::trace_span!("storage.reader.page"));
84        let future = async {
85            let Some(payload_limit) = self.snapshot_bytes else {
86                return Ok(None);
87            };
88            let encoded_bound =
89                recovery_encoded_bound(payload_limit.get(), self.storage.inner.recovery.history(), self.storage.inner.primary.history())?;
90            let encoded_bound = NonZeroUsize::new(encoded_bound).expect("recovery envelope is nonempty");
91            let snapshot =
92                self.storage.inner.bounded("recovery.read_snapshot", self.storage.inner.recovery.read_snapshot(encoded_bound)).await?;
93            if let Some(snapshot) = &snapshot
94                && (!self.through.covers(&snapshot.through)? || snapshot.payload.len() > payload_limit.get())
95            {
96                return Err(Error::InvalidInput {
97                    field: "state snapshot",
98                    reason: "invalid managed snapshot boundary or size",
99                });
100            }
101            if let Some(snapshot) = &snapshot {
102                self.storage.check_state_dependency(&snapshot.required_data)?;
103            }
104            Ok(snapshot.map(crate::ReadBatch::new))
105        };
106        context.scope(tracing::Instrument::instrument(future, span)).await
107    }
108
109    pub async fn page(&self, after: &RecoveryPosition) -> Result<crate::ReadBatch<RecoveryPage>> {
110        let context = self.context.clone();
111        let span = context.sync(|| tracing::trace_span!("storage.reader.page"));
112        let future = async {
113            let page = self
114                .storage
115                .inner
116                .bounded("recovery.read_entries", self.storage.inner.recovery.read_entries(after, &self.through, self.limits))
117                .await?;
118            page.check(after, &self.through, self.limits, |entry| encoded_size(entry).unwrap_or(usize::MAX))?;
119            for entry in &page.entries {
120                self.storage.check_state_dependency(&entry.required_data)?;
121            }
122            Ok(crate::ReadBatch::new(page))
123        };
124        context.scope(tracing::Instrument::instrument(future, span)).await
125    }
126
127    pub fn finish(self, position: Position) -> watch::Receiver<Position> {
128        let (saved, durable) = watch::channel(position);
129        let mut current = self.storage.inner.state.lock().unwrap();
130        current.journal_required_data = current.status.saved.clone();
131        current.journal_saved = Some(saved);
132        current.journal_limit = self.limits.bytes.get();
133        durable
134    }
135}
136
137/* === Public Functions === */
138
139/* === Private Functions === */
140
141/* === Tests === */
harmos_storage/state/reader.rs

harmos_storage/state/
reader.rs
