harmos_storage/manager/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod accounting;
13mod admission;
14mod arrow;
15mod encoding;
16mod lifecycle;
17mod recovery;
18mod shutdown;
19mod state;
20mod status;
21mod writes;
22
23/* === Re-Exports === */
24
25pub use encoding::encoded_size;
26pub(crate) use encoding::{recovery_preparation_bytes, recovery_resident_bytes};
27
28/* === Imports === */
29
30// Standard Library Imports
31use std::collections::BTreeMap;
32use std::io::{self, Write};
33use std::marker::PhantomData;
34use std::sync::{Arc, Mutex};
35
36// Third Party Imports
37use serde::Serialize;
38use tokio::sync::watch;
39
40// Local Imports
41use super::admissions::Mutation;
42use super::arrow::RetainedMemory;
43use super::capacity::{Attempt, Held, Pool, Refusal, Reservation};
44use super::config::{IO_TIMEOUT, MUTATION_RETENTION};
45use super::queue::PendingQueue;
46use super::receipts::{MutationValue, ReceiptState};
47use super::state::{RecoveryEntry, RecoveryPage, RecoverySnapshot, StateBackend};
48use super::worker::Worker;
49use super::{
50    ReadLimits, RecoveryPosition, SchemaKey, StorageBackend, StorageBackends, StorageCursor, StorageFuture, StorageSchema, StorageSettings,
51    StoredRecord, StoredRow,
52};
53use crate::execution::Execution;
54use crate::{Error, Identifier, Result};
55use harmos_domain::model::storage::{RecordWrite, StreamWrite};
56
57/* === Constants === */
58
59const WORKER_NAME: &str = "harmos-storage";
60
61// Includes owned envelopes, queue nodes, Arc headers, and cloned metadata.
62pub(super) const ENTRY_OVERHEAD: usize = 1024;
63
64/* === Macros === */
65
66/* === Type Aliases === */
67
68/* === Structs === */
69
70#[derive(Clone, Debug)]
71pub struct StorageMetrics {
72    pub accepted: StorageCursor,
73    pub saved: StorageCursor,
74    pub recovery_accepted: RecoveryPosition,
75    pub recovery_saved: RecoveryPosition,
76    pub pending_bytes: usize,
77    pub read_bytes: usize,
78    pub mutation_bytes: usize,
79    pub pending_batches: usize,
80    pub running: bool,
81    pub closing: bool,
82    pub closed: bool,
83    pub failure: Option<Arc<Error>>,
84}
85
86pub struct StorageManager<C = ()> {
87    pub(super) inner: Arc<Inner>,
88    catalog: PhantomData<fn() -> C>,
89}
90
91pub(super) struct Inner {
92    pub(super) execution: Execution,
93    pub(super) measurement_admission: tokio::sync::Mutex<()>,
94    pub(super) control_admission: tokio::sync::Mutex<()>,
95    measurement_waiting: std::sync::atomic::AtomicUsize,
96    control_waiting: std::sync::atomic::AtomicUsize,
97    pub(super) object_writes: tokio::sync::Mutex<()>,
98    pub(super) query_active: Arc<tokio::sync::Semaphore>,
99    pub(super) query_pending: Arc<tokio::sync::Semaphore>,
100    pub(super) settings: StorageSettings,
101    pub(super) primary: Arc<dyn StorageBackend>,
102    pub(super) recovery: Arc<dyn StateBackend>,
103    pub(super) has_objects: bool,
104    pub(super) object_definition: fn(&SchemaKey) -> harmos_domain::Result<harmos_domain::model::storage::catalog::ObjectDefinition>,
105    pub(super) state: Mutex<State>,
106    pub(super) changed: watch::Sender<u64>,
107    pub(super) lifecycle: tokio::sync::RwLock<Lifecycle>,
108}
109
110#[derive(Default)]
111pub(super) struct Lifecycle {
112    startup_complete: bool,
113    primary_touched: bool,
114    recovery_touched: bool,
115    primary_open: bool,
116    files_recovered: bool,
117    receipts_recovered: bool,
118    recovery_open: bool,
119    primary_closed: bool,
120    recovery_closed: bool,
121    owner: Option<Box<dyn WorkerOwner>>,
122}
123
124pub(super) struct State {
125    pub(super) execution: Execution,
126    prefer_measurement: bool,
127    pub(super) journal_required_data: StorageCursor,
128    pub(super) reserved: usize,
129    pub(super) reserved_batches: usize,
130    pub(super) journal_limit: usize,
131    pub(super) journal_saved: Option<watch::Sender<harmos_domain::state::Position>>,
132    pub(super) status: StorageMetrics,
133    pub(super) ready: bool,
134    service_owner: Option<Arc<()>>,
135    pub(super) next: u64,
136    pub(super) queue: PendingQueue,
137    pub(super) scratch_bytes: usize,
138    pub(super) read_bytes: usize,
139
140    pub(super) growers: [super::capacity::Growers; 2],
141
142    pub(super) data_attempted: u64,
143    pub(super) objects_pruned: u64,
144    pub(super) objects_prune_owed: bool,
145    pub(super) pruning_objects: bool,
146    pub(super) recovery_written: Target,
147    pub(super) data_accepted: Target,
148    pub(super) recovery_accepted: Target,
149    pub(super) data_flushed: u64,
150    pub(super) recovery_flushed: u64,
151    pub(super) data_request: Option<Target>,
152    pub(super) recovery_request: Option<Target>,
153    pub(super) retry: u64,
154    pub(super) snapshots: BTreeMap<u64, super::snapshot::SnapshotState>,
155    pub(super) followers: BTreeMap<u64, super::follow::Following>,
156    pub(super) files: BTreeMap<u64, super::staging::FileState>,
157    pub(super) receipts: super::receipts::Receipts,
158    pub(super) deletions: BTreeMap<super::MutationId, super::deletion::Deletion>,
159    pub(super) file_cursor: u64,
160    pub(super) finishing: Option<u64>,
161}
162
163#[derive(Clone, Copy, Debug, Default)]
164pub(super) struct Target {
165    pub(super) index: u64,
166    pub(super) sequence: u64,
167}
168
169pub(super) struct Pending {
170    pub(super) journal: Option<harmos_domain::state::Position>,
171    pub(super) index: u64,
172    pub(super) bytes: usize,
173    pub(super) operation: Operation,
174}
175
176struct Counter(usize);
177
178struct Admission {
179    inner: Arc<Inner>,
180    active: bool,
181}
182
183struct WriteWaiter {
184    inner: Arc<Inner>,
185    measurement: bool,
186}
187
188/* === Enums === */
189
190enum Admitted {
191    Replayed(u64),
192    Reserved(u64, u64, u64),
193}
194
195pub enum ErasedCatalog {}
196
197#[derive(Clone, Copy, PartialEq, Eq)]
198pub(super) enum Domain {
199    Data,
200    Recovery,
201}
202
203pub(super) enum Operation {
204    Schema(StorageSchema),
205    File(super::FileReference),
206    Records(Vec<StoredRecord>),
207    ObjectGroup { schemas: Vec<StorageSchema>, objects: Vec<super::ObjectEntry>, bytes: usize },
208    Stream(Vec<StoredRow>),
209    Recovery(Vec<RecoveryEntry>),
210    Snapshot(RecoverySnapshot),
211}
212
213/* === Traits === */
214
215pub(crate) trait RecordWriteBytes {
216    fn resident_bytes(&self, history: &super::StoreHistory) -> Result<usize>;
217}
218
219pub trait WorkerOwner: Send + Sync {
220    fn stop(&self) -> StorageFuture<'_, ()>;
221}
222
223/* === Implementations === */
224
225impl<C> Clone for StorageManager<C> {
226    fn clone(&self) -> Self {
227        Self { inner: self.inner.clone(), catalog: PhantomData }
228    }
229}
230
231impl<C> StorageManager<C> {
232    pub fn erased(&self) -> StorageManager<ErasedCatalog> {
233        StorageManager { inner: self.inner.clone(), catalog: PhantomData }
234    }
235}
236
237impl<C: super::ObjectCatalog> StorageManager<C> {
238    pub fn new(backends: StorageBackends, settings: StorageSettings) -> Result<Self> {
239        Self::with_execution(backends, settings, Execution::default())
240    }
241
242    pub fn with_execution(backends: StorageBackends, settings: StorageSettings, execution: Execution) -> Result<Self> {
243        C::validate()?;
244        settings.validate()?;
245        let StorageBackends { primary, state: state_backend } = backends;
246        primary.history().check(state_backend.history())?;
247        let history = primary.history().clone();
248        let position = StorageCursor { history: history.clone(), sequence: 0 };
249        let recovery_position = RecoveryPosition { history, sequence: 0 };
250        let receipts = super::receipts::Receipts::new(MUTATION_RETENTION, settings.mutation_bytes(), execution)?;
251        Ok(Self {
252            catalog: PhantomData,
253            inner: Arc::new(Inner {
254                execution,
255                measurement_admission: tokio::sync::Mutex::new(()),
256                control_admission: tokio::sync::Mutex::new(()),
257                measurement_waiting: std::sync::atomic::AtomicUsize::new(0),
258                control_waiting: std::sync::atomic::AtomicUsize::new(0),
259                object_writes: tokio::sync::Mutex::new(()),
260                query_active: Arc::new(tokio::sync::Semaphore::new(settings.request.max_active_queries.get())),
261                query_pending: Arc::new(tokio::sync::Semaphore::new(settings.request.max_pending_queries.get())),
262                settings,
263                primary,
264                recovery: state_backend,
265                has_objects: C::definition_count() != 0,
266                object_definition: harmos_domain::model::storage::catalog::definition::<C>,
267                state: Mutex::new(State {
268                    execution,
269                    prefer_measurement: false,
270                    journal_required_data: position.clone(),
271                    reserved: 0,
272                    reserved_batches: 0,
273                    journal_limit: 0,
274                    journal_saved: None,
275                    status: StorageMetrics {
276                        accepted: position.clone(),
277                        saved: position,
278                        recovery_accepted: recovery_position.clone(),
279                        recovery_saved: recovery_position,
280                        pending_bytes: 0,
281                        read_bytes: 0,
282                        mutation_bytes: 0,
283                        pending_batches: 0,
284                        running: false,
285                        closing: false,
286                        closed: false,
287                        failure: None,
288                    },
289                    ready: false,
290                    service_owner: None,
291                    next: 0,
292                    queue: PendingQueue::default(),
293                    scratch_bytes: 0,
294                    growers: Default::default(),
295                    read_bytes: 0,
296                    data_attempted: 0,
297                    objects_pruned: 0,
298                    objects_prune_owed: false,
299                    pruning_objects: false,
300                    recovery_written: Target::default(),
301                    data_accepted: Target::default(),
302                    recovery_accepted: Target::default(),
303                    data_flushed: 0,
304                    recovery_flushed: 0,
305                    data_request: None,
306                    recovery_request: None,
307                    retry: 0,
308                    snapshots: BTreeMap::new(),
309                    followers: BTreeMap::new(),
310                    files: BTreeMap::new(),
311                    deletions: BTreeMap::new(),
312                    receipts,
313                    file_cursor: 0,
314                    finishing: None,
315                }),
316                changed: watch::channel(0).0,
317                lifecycle: tokio::sync::RwLock::new(Lifecycle::default()),
318            }),
319        })
320    }
321}
322
323/* === Public Functions === */
324
325/* === Private Functions === */
326
327/* === Tests === */
harmos_storage/manager/mod.rs

harmos_storage/manager/
mod.rs
