harmos-storage 0.1.3

Harmos storage: durable journals, snapshots and Arrow-backed stream storage.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports

// Third Party Imports

// Local Imports
use super::*;

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl<C> StorageManager<C> {
    pub fn check_state_dependency(&self, required: &StorageCursor) -> Result<()> {
        if !self.cursor_saved().covers(required)? {
            return Err(Error::InvalidInput { field: "state evidence dependency", reason: "required primary evidence is not durable" });
        }
        Ok(())
    }

    pub fn retry_transient_persistence(&self) -> bool {
        let mut state = self.inner.state.lock().unwrap();
        let Some(failure) = &state.status.failure else {
            return false;
        };
        let mut cause = failure.as_ref();
        while let Error::Persistence { source } = cause {
            cause = source.as_ref();
        }
        if !matches!(cause, Error::Backend { kind: crate::BackendFailure::Transient, .. }) {
            return false;
        }
        state.status.failure = None;
        state.retry = state.retry.wrapping_add(1);
        drop(state);
        self.inner.notify();
        true
    }

    pub fn reserve_arrow_memory(&self, bytes: usize) -> Result<RetainedMemory> {
        Reservation::try_new(&self.inner, Pool::Read, &Held::default(), bytes).map(Reservation::retain).map_err(Refusal::into_error)
    }

    pub fn write_receipt_capacity(&self) -> usize {
        self.inner.settings.request.max_pending_write_batches.get()
    }

    pub async fn await_durable(&self, cursor: &StorageCursor) -> Result<()> {
        self.inner.primary.history().check(&cursor.history)?;
        self.flush_domain(Domain::Data, Some(cursor)).await
    }

    pub async fn release_object_history(&self, schema: &SchemaKey, item: &crate::ObjectKey, through: &StorageCursor) -> Result<()> {
        self.inner.primary.history().check(&through.history)?;
        if !self.status().saved.covers(through)? {
            return Err(Error::InvalidInput { field: "object retention", reason: "release exceeds durable frontier" });
        }
        self.inner.bounded(self.inner.primary.release_object_history(schema, item, through)).await?;
        self.inner.state.lock().unwrap().objects_pruned = 0;
        self.inner.notify();
        Ok(())
    }
}

/* === Public Functions === */

/* === Private Functions === */

/* === Tests === */