harmos_storage/
backend.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Storage Backend Contract
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::future::Future;
20use std::num::NonZeroUsize;
21use std::pin::Pin;
22use std::sync::Arc;
23
24// Third Party Imports
25
26// Local Imports
27use super::query::{FileSelection, ObjectSelection, RecordSelection, SchemaSelection, StreamSelection};
28use super::{
29    FileCheckpoint, FileMetadata, FileProgress, FileReference, MutationReceipt, ObjectEntry, ObjectKey, ObjectPoint, SchemaKey, StagedFile,
30    StorageBatch, StorageCursor, StorageSchema, StoreHistory, StoredRecord, StoredRow,
31};
32use crate::{Error, Result};
33use harmos_domain::model::storage::{IdentitySelection, StoredIdentity};
34
35/* === Constants === */
36
37/* === Macros === */
38
39/* === Type Aliases === */
40
41pub type StorageFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T>> + Send + 'a>>;
42
43/* === Structs === */
44
45pub struct StorageBackends {
46    pub primary: Arc<dyn StorageBackend>,
47    pub state: Arc<dyn super::state::StateBackend>,
48}
49
50#[derive(Clone, Debug, PartialEq, Eq)]
51pub struct BackendSnapshot {
52    pub at: StorageCursor,
53    pub token: Vec<u8>,
54}
55
56#[derive(Clone, Copy, Debug, PartialEq, Eq)]
57pub struct ScanPage {
58    pub after: u64,
59    pub rows: NonZeroUsize,
60    pub bytes: NonZeroUsize,
61}
62
63#[derive(Clone, Copy, Debug, PartialEq, Eq)]
64pub struct ReadLimits {
65    pub rows: NonZeroUsize,
66    pub bytes: NonZeroUsize,
67}
68
69/* === Enums === */
70
71/* === Traits === */
72
73pub trait StorageBackend: Send + Sync + 'static {
74    // Lifecycle
75    fn history(&self) -> &StoreHistory;
76    fn open(&self) -> StorageFuture<'_, StorageCursor>;
77    fn cursor_saved(&self) -> StorageFuture<'_, StorageCursor>;
78    fn close(&self) -> StorageFuture<'_, ()>;
79
80    // Atomic Durable Writes
81    fn write<'a>(&'a self, batch: StorageBatch<'a>) -> StorageFuture<'a, StorageCursor>;
82
83    // Historical Reads
84    fn open_snapshot<'a>(&'a self, at: &'a StorageCursor) -> StorageFuture<'a, BackendSnapshot>;
85    fn release_snapshot<'a>(&'a self, snapshot: &'a BackendSnapshot) -> StorageFuture<'a, ()>;
86    fn read_schemas<'a>(
87        &'a self, snapshot: &'a BackendSnapshot, selection: &'a SchemaSelection, page: ScanPage,
88    ) -> StorageFuture<'a, Vec<StorageSchema>>;
89    fn read_objects<'a>(
90        &'a self, snapshot: &'a BackendSnapshot, selection: &'a ObjectSelection, page: ScanPage,
91    ) -> StorageFuture<'a, Vec<ObjectEntry>>;
92    fn read_records<'a>(
93        &'a self, snapshot: &'a BackendSnapshot, selection: &'a RecordSelection, page: ScanPage,
94    ) -> StorageFuture<'a, Vec<StoredRecord>>;
95    fn read_stream<'a>(
96        &'a self, snapshot: &'a BackendSnapshot, selection: &'a StreamSelection, page: ScanPage,
97    ) -> StorageFuture<'a, Vec<StoredRow>>;
98    fn read_files<'a>(
99        &'a self, snapshot: &'a BackendSnapshot, selection: &'a FileSelection, page: ScanPage,
100    ) -> StorageFuture<'a, Vec<FileReference>>;
101    fn read_object<'a>(&'a self, snapshot: &'a BackendSnapshot, point: &'a ObjectPoint) -> StorageFuture<'a, Option<ObjectEntry>>;
102    fn read_file<'a>(
103        &'a self, snapshot: &'a BackendSnapshot, file: &'a FileReference, offset: u64, max_bytes: NonZeroUsize,
104    ) -> StorageFuture<'a, Vec<u8>>;
105
106    // Object Lifecycle Retention
107    fn release_object_history<'a>(
108        &'a self, schema: &'a SchemaKey, item: &'a ObjectKey, through: &'a StorageCursor,
109    ) -> StorageFuture<'a, ()>;
110
111    fn prune_objects<'a>(&'a self, through: &'a StorageCursor, max_entries: NonZeroUsize) -> StorageFuture<'a, bool>;
112
113    // Finished File Publication
114    fn create_file<'a>(&'a self, metadata: &'a FileMetadata) -> StorageFuture<'a, StagedFile>;
115    fn write_file<'a>(&'a self, staging: &'a StagedFile, offset: u64, bytes: &'a [u8]) -> StorageFuture<'a, ()>;
116    fn finish_file<'a>(
117        &'a self, staging: &'a StagedFile, position: &'a StorageCursor, length: u64, sha256: [u8; 32],
118    ) -> StorageFuture<'a, FileReference>;
119    fn abort_file<'a>(&'a self, staging: &'a StagedFile) -> StorageFuture<'a, ()>;
120
121    // Resumable File Sessions
122    fn supports_file_sessions(&self) -> bool {
123        false
124    }
125    fn create_production<'a>(&'a self, _metadata: &'a FileMetadata) -> StorageFuture<'a, StagedFile> {
126        unsupported("file session")
127    }
128    fn fence_production<'a>(&'a self, _file: &'a StagedFile, _expected: u64, _next: u64) -> StorageFuture<'a, ()> {
129        unsupported("file session")
130    }
131    fn save_production<'a>(&'a self, _checkpoint: &'a FileCheckpoint) -> StorageFuture<'a, ()> {
132        unsupported("file checkpoint")
133    }
134    fn recover_productions(&self, _limits: ReadLimits) -> StorageFuture<'_, Vec<FileCheckpoint>> {
135        unsupported("file recovery")
136    }
137    fn file_progress<'a>(&'a self, _file: &'a StagedFile) -> StorageFuture<'a, FileProgress> {
138        unsupported("file following")
139    }
140    fn read_production<'a>(&'a self, _file: &'a StagedFile, _offset: u64, _max_bytes: NonZeroUsize) -> StorageFuture<'a, Vec<u8>> {
141        unsupported("file following")
142    }
143    fn abandon_file<'a>(&'a self, _staging: &'a StagedFile) -> StorageFuture<'a, ()> {
144        unsupported("file session")
145    }
146    fn supports_file_deletion(&self) -> bool {
147        false
148    }
149    fn delete_file<'a>(&'a self, _file: &'a StagedFile) -> StorageFuture<'a, ()> {
150        unsupported("file deletion")
151    }
152
153    // Idempotent Mutation Receipts
154    fn supports_mutation_receipts(&self) -> bool {
155        false
156    }
157    fn save_mutations<'a>(&'a self, _receipts: &'a [MutationReceipt]) -> StorageFuture<'a, ()> {
158        unsupported("storage mutations")
159    }
160
161    fn recover_mutations(&self, _limits: ReadLimits) -> StorageFuture<'_, Vec<MutationReceipt>> {
162        unsupported("storage mutations")
163    }
164
165    // Stored Identity Catalog
166    fn read_identities<'a>(
167        &'a self, _snapshot: &'a BackendSnapshot, _selection: &'a IdentitySelection, _page: ScanPage,
168    ) -> StorageFuture<'a, Vec<StoredIdentity>> {
169        unsupported("storage identity catalog")
170    }
171}
172
173/* === Implementations === */
174
175impl ReadLimits {
176    pub fn check(&self, rows: usize, bytes: usize) -> Result<()> {
177        if rows > self.rows.get() || bytes > self.bytes.get() {
178            return Err(Error::InvalidInput { field: "read page", reason: "row or byte bound exceeded" });
179        }
180        Ok(())
181    }
182}
183
184/* === Public Functions === */
185
186/* === Private Functions === */
187
188fn unsupported<'a, T>(field: &'static str) -> StorageFuture<'a, T> {
189    Box::pin(async move { Err(Error::InvalidInput { field, reason: "backend does not support this capability" }) })
190}
191
192/* === Tests === */
harmos_storage/backend.rs

harmos_storage/
backend.rs
