harmos-storage 0.1.5

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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

/* ***
 *
 * 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 RecordWriteBytes for RecordWrite<'_> {
    fn resident_bytes(&self, history: &crate::StoreHistory) -> Result<usize> {
        [
            self.payload.len(),
            self.schema.id.as_str().len(),
            self.scope.map_or(0, |value| value.as_str().len()),
            self.item.as_str().len(),
            self.source.map_or(0, |value| value.as_str().len()),
            history.store.as_str().len(),
            history.epoch.as_str().len(),
            std::mem::size_of::<StoredRecord>(),
            ENTRY_OVERHEAD,
        ]
        .into_iter()
        .try_fold(0usize, |bytes, next| bytes.checked_add(next))
        .ok_or(Error::InvalidInput { field: "storage admission", reason: "size overflow" })
    }
}

impl Drop for WriteWaiter {
    fn drop(&mut self) {
        let counter = if self.measurement { &self.inner.measurement_waiting } else { &self.inner.control_waiting };
        counter.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
        self.inner.notify();
    }
}

impl Drop for Admission {
    fn drop(&mut self) {
        if self.active {
            {
                let mut state = self.inner.state.lock().unwrap();
                state.reserved = 0;
                state.reserved_batches = 0;
                state.collect();
            }
            self.inner.notify();
        }
    }
}

impl Write for Counter {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.0 = self.0.checked_add(bytes.len()).ok_or_else(|| io::Error::other("encoded size overflow"))?;
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

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

pub fn encoded_size<T: Serialize + ?Sized>(value: &T) -> Result<usize> {
    let mut counter = Counter(0);
    serde_json::to_writer(&mut counter, value)
        .map_err(|_| Error::InvalidInput { field: "storage payload", reason: "cannot measure encoded size" })?;
    Ok(counter.0)
}

pub(crate) fn recovery_preparation_bytes(
    payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory,
) -> Result<usize> {
    let capacity = payload
        .checked_mul(4)
        .and_then(|bytes| bytes.checked_add(128))
        .ok_or(Error::InvalidInput { field: "managed recovery", reason: "size overflow" })?;
    recovery_resident_bytes(capacity, history, data_history)
}

pub(crate) fn recovery_resident_bytes(payload: usize, history: &crate::StoreHistory, data_history: &crate::StoreHistory) -> Result<usize> {
    [
        payload,
        history.store.as_str().len(),
        history.epoch.as_str().len(),
        data_history.store.as_str().len(),
        data_history.epoch.as_str().len(),
        std::mem::size_of::<RecoveryEntry>(),
        ENTRY_OVERHEAD,
    ]
    .into_iter()
    .try_fold(0usize, |bytes, next| bytes.checked_add(next))
    .ok_or(Error::InvalidInput { field: "managed recovery", reason: "size overflow" })
}

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

/* === Tests === */