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

harmos_storage/manager/
mod.rs
