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, StorageBatch, StorageCursor, StorageFuture, StorageSchema,
51    StorageSettings, StoredRecord, StoredRow,
52};
53use crate::execution::Execution;
54use crate::{Error, Identifier, Result};
55use harmos_domain::model::storage::{RecordInput, RecordWrite, StreamInput};
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) retained_write_bytes: Arc<std::sync::atomic::AtomicUsize>,
94    pub(super) retained_read_bytes: Arc<std::sync::atomic::AtomicUsize>,
95    pub(super) measurement_admission: tokio::sync::Mutex<()>,
96    pub(super) control_admission: tokio::sync::Mutex<()>,
97    measurement_waiting: std::sync::atomic::AtomicUsize,
98    control_waiting: std::sync::atomic::AtomicUsize,
99    pub(super) object_writes: tokio::sync::Mutex<()>,
100    pub(super) query_active: Arc<tokio::sync::Semaphore>,
101    pub(super) query_pending: Arc<tokio::sync::Semaphore>,
102    pub(super) settings: StorageSettings,
103    pub(super) primary: Arc<dyn StorageBackend>,
104    pub(super) recovery: Arc<dyn StateBackend>,
105    pub(super) has_objects: bool,
106    pub(super) object_definition: fn(&SchemaKey) -> harmos_domain::Result<harmos_domain::model::storage::catalog::ObjectDefinition>,
107    pub(super) object_retention: fn(&SchemaKey) -> harmos_domain::Result<super::ObjectRetention>,
108    pub(super) state: Mutex<State>,
109    pub(super) changed: watch::Sender<u64>,
110    pub(super) lifecycle: tokio::sync::RwLock<Lifecycle>,
111}
112
113#[derive(Default)]
114pub(super) struct Lifecycle {
115    startup_complete: bool,
116    primary_touched: bool,
117    recovery_touched: bool,
118    primary_open: bool,
119    files_recovered: bool,
120    receipts_recovered: bool,
121    recovery_open: bool,
122    primary_closed: bool,
123    recovery_closed: bool,
124    owner: Option<Box<dyn WorkerOwner>>,
125}
126
127pub(super) struct State {
128    pub(super) execution: Execution,
129    pub(super) retained_write_bytes: Arc<std::sync::atomic::AtomicUsize>,
130    pub(super) retained_read_bytes: Arc<std::sync::atomic::AtomicUsize>,
131    prefer_measurement: bool,
132    pub(super) journal_required_data: StorageCursor,
133    pub(super) reserved: usize,
134    pub(super) reserved_batches: usize,
135    pub(super) journal_limit: usize,
136    pub(super) journal_saved: Option<watch::Sender<harmos_domain::state::Position>>,
137    pub(super) status: StorageMetrics,
138    pub(super) ready: bool,
139    service_owner: Option<Arc<()>>,
140    pub(super) next: u64,
141    pub(super) queue: PendingQueue,
142    pub(super) scratch_bytes: usize,
143    pub(super) read_bytes: usize,
144
145    pub(super) growers: [super::capacity::Growers; 2],
146
147    pub(super) data_attempted: u64,
148    pub(super) objects_pruned: u64,
149    pub(super) objects_prune_owed: bool,
150    pub(super) pruning_objects: bool,
151    pub(super) recovery_written: Target,
152    pub(super) data_accepted: Target,
153    pub(super) recovery_accepted: Target,
154    pub(super) data_flushed: u64,
155    pub(super) recovery_flushed: u64,
156    pub(super) data_request: Option<Target>,
157    pub(super) recovery_request: Option<Target>,
158    pub(super) retry: u64,
159    pub(super) snapshots: BTreeMap<u64, super::snapshot::SnapshotState>,
160    pub(super) followers: BTreeMap<u64, super::follow::Following>,
161    pub(super) files: BTreeMap<u64, super::staging::FileState>,
162    pub(super) receipts: super::receipts::Receipts,
163    pub(super) deletions: BTreeMap<super::MutationId, super::deletion::Deletion>,
164    pub(super) file_cursor: u64,
165    pub(super) finishing: Option<u64>,
166}
167
168#[derive(Clone, Copy, Debug, Default)]
169pub(super) struct Target {
170    pub(super) index: u64,
171    pub(super) sequence: u64,
172}
173
174pub(super) struct Pending {
175    pub(super) context: crate::execution::Context,
176    pub(super) queued: Mutex<Option<tracing::Span>>,
177    pub(super) journal: Option<harmos_domain::state::Position>,
178    pub(super) index: u64,
179    pub(super) bytes: usize,
180    pub(super) operation: Operation,
181    charge: Option<Arc<crate::batch::WriteCharge>>,
182}
183
184struct Counter(usize);
185
186struct Admission {
187    inner: Arc<Inner>,
188    active: bool,
189}
190
191struct WriteWaiter {
192    inner: Arc<Inner>,
193    measurement: bool,
194}
195
196/* === Enums === */
197
198enum Admitted {
199    Replayed(u64),
200    Reserved(u64, u64, u64),
201}
202
203pub enum ErasedCatalog {}
204
205#[derive(Clone, Copy, PartialEq, Eq)]
206pub(super) enum Domain {
207    Data,
208    Recovery,
209}
210
211pub(super) enum Operation {
212    Schema(StorageBatch<StorageSchema>),
213    File(StorageBatch<super::FileReference>),
214    Records(StorageBatch<StoredRecord>),
215    ObjectGroup { schemas: StorageBatch<StorageSchema>, objects: StorageBatch<super::ObjectEntry>, bytes: usize },
216    Stream(StorageBatch<StoredRow>),
217    Recovery(StorageBatch<RecoveryEntry>),
218    Snapshot(StorageBatch<RecoverySnapshot>),
219}
220
221/* === Traits === */
222
223pub(crate) trait RecordWriteBytes {
224    fn resident_bytes(&self, history: &super::StoreHistory) -> Result<usize>;
225}
226
227pub trait WorkerOwner: Send + Sync {
228    fn stop(&self) -> StorageFuture<'_, ()>;
229}
230
231/* === Implementations === */
232
233impl Pending {
234    pub(super) fn retire(&self) {
235        if let Some(charge) = &self.charge {
236            charge.retire();
237        }
238    }
239
240    pub(super) fn new(
241        inner: &Arc<Inner>, journal: Option<harmos_domain::state::Position>, index: u64, bytes: usize, mut operation: Operation,
242    ) -> Self {
243        let charge = match &mut operation {
244            Operation::Schema(values) => Some({
245                let charge = crate::batch::WriteCharge::new(inner, bytes);
246                *values = values.clone().retained(Arc::clone(&charge));
247                charge
248            }),
249            Operation::File(values) => Some({
250                let charge = crate::batch::WriteCharge::new(inner, bytes);
251                *values = values.clone().retained(Arc::clone(&charge));
252                charge
253            }),
254            Operation::Records(values) => Some({
255                let charge = crate::batch::WriteCharge::new(inner, bytes);
256                *values = values.clone().retained(Arc::clone(&charge));
257                charge
258            }),
259            Operation::Stream(values) => Some({
260                let charge = crate::batch::WriteCharge::new(inner, bytes);
261                *values = values.clone().retained(Arc::clone(&charge));
262                charge
263            }),
264            Operation::ObjectGroup { schemas, objects, .. } => Some({
265                let charge = crate::batch::WriteCharge::new(inner, bytes);
266                *schemas = schemas.clone().retained(Arc::clone(&charge));
267                *objects = objects.clone().retained(Arc::clone(&charge));
268                charge
269            }),
270            Operation::Recovery(values) => Some({
271                let charge = crate::batch::WriteCharge::new(inner, bytes);
272                *values = values.clone().retained(Arc::clone(&charge));
273                charge
274            }),
275            Operation::Snapshot(values) => Some({
276                let charge = crate::batch::WriteCharge::new(inner, bytes);
277                *values = values.clone().retained(Arc::clone(&charge));
278                charge
279            }),
280        };
281        Self {
282            context: crate::execution::Context::current(),
283            queued: Mutex::new(Some(tracing::debug_span!("storage.queue"))),
284            journal,
285            index,
286            bytes,
287            operation,
288            charge,
289        }
290    }
291}
292
293impl Drop for Pending {
294    fn drop(&mut self) {
295        self.retire();
296        self.context.sync(|| drop(self.queued.get_mut().unwrap().take()));
297    }
298}
299
300impl<C> Clone for StorageManager<C> {
301    fn clone(&self) -> Self {
302        Self { inner: self.inner.clone(), catalog: PhantomData }
303    }
304}
305
306impl<C> StorageManager<C> {
307    pub fn erased(&self) -> StorageManager<ErasedCatalog> {
308        StorageManager { inner: self.inner.clone(), catalog: PhantomData }
309    }
310}
311
312impl<C: super::ObjectCatalog> StorageManager<C> {
313    pub fn new(backends: StorageBackends, settings: StorageSettings) -> Result<Self> {
314        Self::with_execution(backends, settings, Execution::default())
315    }
316
317    #[tracing::instrument(name = "storage.manager.new", level = "debug", skip_all)]
318    pub fn with_execution(backends: StorageBackends, settings: StorageSettings, execution: Execution) -> Result<Self> {
319        C::validate()?;
320        settings.validate()?;
321        let StorageBackends { primary, state: state_backend } = backends;
322        primary.history().check(state_backend.history())?;
323        let history = primary.history().clone();
324        let position = StorageCursor { history: history.clone(), sequence: 0 };
325        let recovery_position = RecoveryPosition { history, sequence: 0 };
326        let receipts = super::receipts::Receipts::new(MUTATION_RETENTION, settings.mutation_bytes(), execution)?;
327        let retained_write_bytes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
328        let retained_read_bytes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
329        Ok(Self {
330            catalog: PhantomData,
331            inner: Arc::new(Inner {
332                execution,
333                retained_write_bytes: Arc::clone(&retained_write_bytes),
334                retained_read_bytes: Arc::clone(&retained_read_bytes),
335                measurement_admission: tokio::sync::Mutex::new(()),
336                control_admission: tokio::sync::Mutex::new(()),
337                measurement_waiting: std::sync::atomic::AtomicUsize::new(0),
338                control_waiting: std::sync::atomic::AtomicUsize::new(0),
339                object_writes: tokio::sync::Mutex::new(()),
340                query_active: Arc::new(tokio::sync::Semaphore::new(settings.request.max_active_queries.get())),
341                query_pending: Arc::new(tokio::sync::Semaphore::new(settings.request.max_pending_queries.get())),
342                settings,
343                primary,
344                recovery: state_backend,
345                has_objects: C::definition_count() != 0,
346                object_definition: harmos_domain::model::storage::catalog::definition::<C>,
347                object_retention: |key| Ok(harmos_domain::model::storage::catalog::definition::<C>(key)?.retention()),
348                state: Mutex::new(State {
349                    execution,
350                    retained_write_bytes,
351                    retained_read_bytes,
352                    prefer_measurement: false,
353                    journal_required_data: position.clone(),
354                    reserved: 0,
355                    reserved_batches: 0,
356                    journal_limit: 0,
357                    journal_saved: None,
358                    status: StorageMetrics {
359                        accepted: position.clone(),
360                        saved: position,
361                        recovery_accepted: recovery_position.clone(),
362                        recovery_saved: recovery_position,
363                        pending_bytes: 0,
364                        read_bytes: 0,
365                        mutation_bytes: 0,
366                        pending_batches: 0,
367                        running: false,
368                        closing: false,
369                        closed: false,
370                        failure: None,
371                    },
372                    ready: false,
373                    service_owner: None,
374                    next: 0,
375                    queue: PendingQueue::default(),
376                    scratch_bytes: 0,
377                    growers: Default::default(),
378                    read_bytes: 0,
379                    data_attempted: 0,
380                    objects_pruned: 0,
381                    objects_prune_owed: false,
382                    pruning_objects: false,
383                    recovery_written: Target::default(),
384                    data_accepted: Target::default(),
385                    recovery_accepted: Target::default(),
386                    data_flushed: 0,
387                    recovery_flushed: 0,
388                    data_request: None,
389                    recovery_request: None,
390                    retry: 0,
391                    snapshots: BTreeMap::new(),
392                    followers: BTreeMap::new(),
393                    files: BTreeMap::new(),
394                    deletions: BTreeMap::new(),
395                    receipts,
396                    file_cursor: 0,
397                    finishing: None,
398                }),
399                changed: watch::channel(0).0,
400                lifecycle: tokio::sync::RwLock::new(Lifecycle::default()),
401            }),
402        })
403    }
404}
405
406/* === Public Functions === */
407
408/* === Private Functions === */
409
410/* === Tests === */
harmos_storage/manager/mod.rs

harmos_storage/manager/
mod.rs
