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, FileChunk, FileMetadata, FileProgress, FileReference, MutationReceipt, ObjectEntry, ObjectKey, ObjectPoint,
30    ObjectRetention, SchemaKey, StagedFile, 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    // Independent Durable Writes; Replayed Positions Must Be Idempotent.
81    fn write_schemas(&self, batch: StorageBatch<StorageSchema>) -> StorageFuture<'_, StorageCursor>;
82    fn write_objects(&self, batch: StorageBatch<ObjectEntry>) -> StorageFuture<'_, StorageCursor>;
83    fn write_records(&self, batch: StorageBatch<StoredRecord>) -> StorageFuture<'_, StorageCursor>;
84    fn write_stream(&self, batch: StorageBatch<StoredRow>) -> StorageFuture<'_, StorageCursor>;
85    fn write_files(&self, batch: StorageBatch<FileReference>) -> StorageFuture<'_, StorageCursor>;
86
87    // Historical Reads
88    fn open_snapshot<'a>(&'a self, at: &'a StorageCursor) -> StorageFuture<'a, BackendSnapshot>;
89    fn release_snapshot<'a>(&'a self, snapshot: &'a BackendSnapshot) -> StorageFuture<'a, ()>;
90    fn read_schemas<'a>(
91        &'a self, snapshot: &'a BackendSnapshot, selection: &'a SchemaSelection, page: ScanPage,
92    ) -> StorageFuture<'a, Vec<StorageSchema>>;
93    fn read_objects<'a>(
94        &'a self, snapshot: &'a BackendSnapshot, selection: &'a ObjectSelection, page: ScanPage,
95    ) -> StorageFuture<'a, Vec<ObjectEntry>>;
96    fn read_records<'a>(
97        &'a self, snapshot: &'a BackendSnapshot, selection: &'a RecordSelection, page: ScanPage,
98    ) -> StorageFuture<'a, Vec<StoredRecord>>;
99    fn read_stream<'a>(
100        &'a self, snapshot: &'a BackendSnapshot, selection: &'a StreamSelection, page: ScanPage,
101    ) -> StorageFuture<'a, Vec<StoredRow>>;
102    fn read_files<'a>(
103        &'a self, snapshot: &'a BackendSnapshot, selection: &'a FileSelection, page: ScanPage,
104    ) -> StorageFuture<'a, Vec<FileReference>>;
105    fn read_object<'a>(&'a self, snapshot: &'a BackendSnapshot, point: &'a ObjectPoint) -> StorageFuture<'a, Option<ObjectEntry>>;
106    fn read_file<'a>(
107        &'a self, snapshot: &'a BackendSnapshot, file: &'a FileReference, offset: u64, max_bytes: NonZeroUsize,
108    ) -> StorageFuture<'a, Vec<u8>>;
109
110    // Object Lifecycle Retention
111    fn release_object_history<'a>(
112        &'a self, schema: &'a SchemaKey, item: &'a ObjectKey, through: &'a StorageCursor,
113    ) -> StorageFuture<'a, ()>;
114
115    fn prune_objects<'a>(
116        &'a self, through: &'a StorageCursor, max_entries: NonZeroUsize,
117        retention: fn(&SchemaKey) -> harmos_domain::Result<ObjectRetention>,
118    ) -> StorageFuture<'a, bool>;
119
120    // Finished File Publication
121    fn create_file<'a>(&'a self, metadata: &'a FileMetadata) -> StorageFuture<'a, StagedFile>;
122    fn write_file<'a>(&'a self, staging: &'a StagedFile, offset: u64, bytes: FileChunk) -> StorageFuture<'a, ()>;
123    fn finish_file<'a>(
124        &'a self, staging: &'a StagedFile, position: &'a StorageCursor, length: u64, sha256: [u8; 32],
125    ) -> StorageFuture<'a, FileReference>;
126    fn abort_file<'a>(&'a self, staging: &'a StagedFile) -> StorageFuture<'a, ()>;
127
128    // Resumable File Sessions
129    fn supports_file_sessions(&self) -> bool {
130        false
131    }
132    fn create_production<'a>(&'a self, _metadata: &'a FileMetadata) -> StorageFuture<'a, StagedFile> {
133        unsupported("file session")
134    }
135    fn fence_production<'a>(&'a self, _file: &'a StagedFile, _expected: u64, _next: u64) -> StorageFuture<'a, ()> {
136        unsupported("file session")
137    }
138    fn save_production(&self, _checkpoint: StorageBatch<FileCheckpoint>) -> StorageFuture<'_, ()> {
139        unsupported("file checkpoint")
140    }
141    fn recover_productions(&self, _limits: ReadLimits) -> StorageFuture<'_, Vec<FileCheckpoint>> {
142        unsupported("file recovery")
143    }
144    fn file_progress<'a>(&'a self, _file: &'a StagedFile) -> StorageFuture<'a, FileProgress> {
145        unsupported("file following")
146    }
147    fn read_production<'a>(&'a self, _file: &'a StagedFile, _offset: u64, _max_bytes: NonZeroUsize) -> StorageFuture<'a, Vec<u8>> {
148        unsupported("file following")
149    }
150    fn abandon_file<'a>(&'a self, _staging: &'a StagedFile) -> StorageFuture<'a, ()> {
151        unsupported("file session")
152    }
153    fn supports_file_deletion(&self) -> bool {
154        false
155    }
156    fn delete_file<'a>(&'a self, _file: &'a StagedFile) -> StorageFuture<'a, ()> {
157        unsupported("file deletion")
158    }
159
160    // Idempotent Mutation Receipts
161    fn supports_mutation_receipts(&self) -> bool {
162        false
163    }
164    fn save_mutations(&self, _receipts: StorageBatch<MutationReceipt>) -> StorageFuture<'_, ()> {
165        unsupported("storage mutations")
166    }
167
168    fn recover_mutations(&self, _limits: ReadLimits) -> StorageFuture<'_, Vec<MutationReceipt>> {
169        unsupported("storage mutations")
170    }
171
172    // Stored Identity Catalog
173    fn read_identities<'a>(
174        &'a self, _snapshot: &'a BackendSnapshot, _selection: &'a IdentitySelection, _page: ScanPage,
175    ) -> StorageFuture<'a, Vec<StoredIdentity>> {
176        unsupported("storage identity catalog")
177    }
178}
179
180/* === Implementations === */
181
182impl ReadLimits {
183    pub fn check(&self, rows: usize, bytes: usize) -> Result<()> {
184        if rows > self.rows.get() || bytes > self.bytes.get() {
185            return Err(Error::InvalidInput {
186                field: "read page",
187                reason: "row or byte bound exceeded",
188            });
189        }
190        Ok(())
191    }
192}
193
194/* === Public Functions === */
195
196/* === Private Functions === */
197
198fn unsupported<'a, T>(field: &'static str) -> StorageFuture<'a, T> {
199    Box::pin(async move {
200        Err(Error::InvalidInput {
201            field,
202            reason: "backend does not support this capability",
203        })
204    })
205}
206
207/* === Tests === */
harmos_storage/backend.rs

harmos_storage/
backend.rs
