harmos_storage/manager/
status.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    pub fn check_state_dependency(&self, required: &StorageCursor) -> Result<()> {
39        if !self.cursor_saved().covers(required)? {
40            return Err(Error::InvalidInput { field: "state evidence dependency", reason: "required primary evidence is not durable" });
41        }
42        Ok(())
43    }
44
45    pub fn retry_transient_persistence(&self) -> bool {
46        let mut state = self.inner.state.lock().unwrap();
47        let Some(failure) = &state.status.failure else {
48            return false;
49        };
50        let mut cause = failure.as_ref();
51        while let Error::Persistence { source } = cause {
52            cause = source.as_ref();
53        }
54        if !matches!(cause, Error::Backend { kind: crate::BackendFailure::Transient, .. }) {
55            return false;
56        }
57        state.status.failure = None;
58        state.retry = state.retry.wrapping_add(1);
59        drop(state);
60        self.inner.notify();
61        true
62    }
63
64    pub fn reserve_arrow_memory(&self, bytes: usize) -> Result<RetainedMemory> {
65        Reservation::try_new(&self.inner, Pool::Read, &Held::default(), bytes).map(Reservation::retain).map_err(Refusal::into_error)
66    }
67
68    pub fn write_receipt_capacity(&self) -> usize {
69        self.inner.settings.request.max_pending_write_batches.get()
70    }
71
72    pub async fn await_durable(&self, cursor: &StorageCursor) -> Result<()> {
73        self.inner.primary.history().check(&cursor.history)?;
74        self.flush_domain(Domain::Data, Some(cursor)).await
75    }
76
77    pub async fn release_object_history(&self, schema: &SchemaKey, item: &crate::ObjectKey, through: &StorageCursor) -> Result<()> {
78        self.inner.primary.history().check(&through.history)?;
79        if !self.status().saved.covers(through)? {
80            return Err(Error::InvalidInput { field: "object retention", reason: "release exceeds durable frontier" });
81        }
82        self.inner.bounded(self.inner.primary.release_object_history(schema, item, through)).await?;
83        self.inner.state.lock().unwrap().objects_pruned = 0;
84        self.inner.notify();
85        Ok(())
86    }
87}
88
89/* === Public Functions === */
90
91/* === Private Functions === */
92
93/* === Tests === */
harmos_storage/manager/status.rs

harmos_storage/manager/
status.rs
