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

harmos_storage/
backend.rs
