harmos_domain/model/storage/
catalog.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::fmt;
18use std::num::{NonZeroU32, NonZeroUsize};
19// Third Party Imports
20// Local Imports
21use crate::catalog::{Catalog, Contains, Member, Projection, Visitor};
22use crate::model::storage::{ObjectRetention, ObjectValue, SchemaKey, check_length, invalid, object_key};
23use crate::{Error, Identified, Identifier, MemorySize, Result, schema};
24
25/* === Constants === */
26
27const MAX_DEFINITIONS: usize = 4096;
28const MAX_IDENTITY_BYTES: usize = 1024;
29
30pub const LOOKUP_MEMORY_BYTES: usize = MAX_IDENTITY_BYTES * 2;
31const VALIDATION_BYTES: usize = 8 * 1024 * 1024;
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37/* === Structs === */
38
39#[derive(Clone, Copy, Debug)]
40pub struct ObjectDefinition {
41    schema: fn() -> Result<SchemaKey>,
42    declaration: schema::declaration::Definition,
43    retention: ObjectRetention,
44    verify: fn(&ObjectValue, NonZeroUsize) -> Result<()>,
45
46    verify_admitted: fn(&ObjectValue, NonZeroUsize, usize) -> Result<()>,
47}
48
49pub struct ObjectDefinitions;
50
51struct KeyMatch<'a> {
52    expected: &'a str,
53    offset: usize,
54}
55
56/* === Enums === */
57
58/* === Traits === */
59
60pub trait StoredObject: Identified + schema::Described + crate::Message + MemorySize + 'static {
61    type Key: fmt::Display + ?Sized;
62
63    const RETENTION: ObjectRetention;
64
65    fn key(&self) -> &Self::Key;
66
67    fn validate(&self) -> Result<()> {
68        Ok(())
69    }
70}
71
72pub trait ObjectCatalog: Catalog<ObjectDefinitions> + Send + Sync + 'static {
73    fn definitions() -> <Self as Catalog<ObjectDefinitions>>::Entries {
74        <Self as Catalog<ObjectDefinitions>>::ENTRIES
75    }
76
77    fn definition_count() -> usize {
78        <<Self as Catalog<ObjectDefinitions>>::Entries as Projection<ObjectDefinition>>::LEN
79    }
80
81    fn validate() -> Result<()> {
82        let count = Self::definition_count();
83        if count > MAX_DEFINITIONS {
84            return Err(invalid("object catalog exceeds its definition limit"));
85        }
86        let entries = Self::definitions();
87        let mut visited = 0usize;
88        entries.try_for_each(|_| {
89            visited += 1;
90            if visited > count {
91                return Err(invalid("object catalog projection count differs"));
92            }
93            Ok(())
94        })?;
95        if visited != count {
96            return Err(invalid("object catalog projection count differs"));
97        }
98        let bytes = count
99            .checked_mul(std::mem::size_of::<(SchemaKey, ObjectRetention)>() + MAX_IDENTITY_BYTES)
100            .and_then(|bytes| bytes.checked_add(MAX_IDENTITY_BYTES + std::mem::size_of::<Vec<(SchemaKey, ObjectRetention)>>()))
101            .ok_or_else(|| invalid("object catalog validation size overflow"))?;
102        if bytes > VALIDATION_BYTES {
103            return Err(invalid("object catalog exceeds its validation byte limit"));
104        }
105        let mut identities = Vec::with_capacity(count);
106        entries.try_for_each(|entry| {
107            if identities.len() >= count {
108                return Err(invalid("object catalog projection count differs"));
109            }
110            let key = entry.schema()?;
111            if key.id.as_str().len() > MAX_IDENTITY_BYTES {
112                return Err(invalid("object schema identity exceeds its byte limit"));
113            }
114            let mut settings = crate::config::SchemaSettings::default();
115            settings.preparation_bytes = NonZeroUsize::new(settings.preparation_bytes.get().min(VALIDATION_BYTES - bytes))
116                .ok_or_else(|| invalid("object schema preparation exceeds allowance"))?;
117            entry.graph(&settings)?;
118            identities.push((key, entry.retention));
119            Ok(())
120        })?;
121        if identities.len() != count {
122            return Err(invalid("object catalog projection count differs"));
123        }
124        identities.sort_unstable_by(|(left, _), (right, _)| (left.id.as_str(), left.version).cmp(&(right.id.as_str(), right.version)));
125        for pair in identities.windows(2) {
126            if pair[0].0.id == pair[1].0.id {
127                if pair[0].1 != pair[1].1 {
128                    return Err(invalid("codec versions disagree on object retention"));
129                }
130                if pair[0].0.version == pair[1].0.version {
131                    return Err(invalid("duplicate object schema identity and version"));
132                }
133            }
134        }
135        Ok(())
136    }
137
138    fn encode<T: StoredObject>(value: &T, max_payload_bytes: NonZeroUsize) -> Result<ObjectValue>
139    where
140        Self: Contains<T>,
141    {
142        Self::validate()?;
143        value.validate()?;
144        let schema = object_schema::<T>()?;
145        let item = object_key(value.key(), max_payload_bytes)?;
146        let length = value.encoded_len();
147        check_length(length, max_payload_bytes)?;
148        let payload = value.encode_to_vec();
149        if payload.len() != length {
150            return Err(invalid("object codec reported an inconsistent length"));
151        }
152        Ok(ObjectValue { schema, item, payload })
153    }
154
155    fn decode<T: StoredObject>(value: &ObjectValue, max_payload_bytes: NonZeroUsize) -> Result<T>
156    where
157        Self: Contains<T>,
158    {
159        Self::validate()?;
160        decode::<T>(value, max_payload_bytes)
161    }
162
163    fn validate_value(value: &ObjectValue, max_payload_bytes: NonZeroUsize) -> Result<()> {
164        Self::validate()?;
165        check_length(value.payload.len(), max_payload_bytes)?;
166        let mut verify = None;
167        Self::definitions().try_for_each(|entry| {
168            if verify.is_none() && entry.schema()? == value.schema {
169                verify = Some(entry.verify);
170            }
171            Ok::<(), Error>(())
172        })?;
173        match verify {
174            Some(verify) => verify(value, max_payload_bytes),
175            None => Err(invalid("object schema is absent from the catalog")),
176        }
177    }
178}
179
180/* === Implementations === */
181
182impl ObjectDefinition {
183    pub fn retention(&self) -> ObjectRetention {
184        self.retention
185    }
186
187    pub fn schema(&self) -> Result<SchemaKey> {
188        (self.schema)()
189    }
190
191    pub fn declaration(&self) -> schema::declaration::Definition {
192        self.declaration
193    }
194
195    pub fn graph(&self, settings: &crate::config::SchemaSettings) -> Result<schema::graph::Schema> {
196        schema::graph::Schema::from_declaration(self.declaration, settings)
197            .map_err(|_| invalid("invalid object schema or preparation allowance"))
198    }
199
200    pub fn validate_admitted(&self, value: &ObjectValue, payload: NonZeroUsize, allowance: usize) -> Result<()> {
201        check_length(value.payload.len(), payload)?;
202        check_length(value.item.as_str().len(), payload)?;
203        (self.verify_admitted)(value, payload, allowance)
204    }
205}
206
207impl Visitor for ObjectDefinitions {
208    type Entry = ObjectDefinition;
209}
210
211impl<T: StoredObject> Member<T> for ObjectDefinitions {
212    const ENTRY: ObjectDefinition = ObjectDefinition {
213        schema: object_schema::<T>,
214        declaration: T::DEFINITION,
215        retention: T::RETENTION,
216        verify: verify::<T>,
217
218        verify_admitted: verify_admitted::<T>,
219    };
220}
221
222impl<C: Catalog<ObjectDefinitions> + Send + Sync + 'static> ObjectCatalog for C {}
223
224impl fmt::Write for KeyMatch<'_> {
225    fn write_str(&mut self, value: &str) -> fmt::Result {
226        let rest = self.expected.get(self.offset..).ok_or(fmt::Error)?;
227        if !rest.starts_with(value) {
228            return Err(fmt::Error);
229        }
230        self.offset += value.len();
231        Ok(())
232    }
233}
234
235/* === Public Functions === */
236
237pub fn definition<C: ObjectCatalog>(key: &SchemaKey) -> Result<ObjectDefinition> {
238    let mut found = None;
239    C::definitions().try_for_each(|entry| {
240        if found.is_none() && entry.schema()? == *key {
241            found = Some(*entry);
242        }
243        Ok::<(), Error>(())
244    })?;
245    found.ok_or_else(|| invalid("object schema is absent from the catalog"))
246}
247
248pub fn object_schema<T: StoredObject>() -> Result<SchemaKey> {
249    let (id, version) = T::definition();
250    if id.as_str().len() > MAX_IDENTITY_BYTES {
251        return Err(invalid("object schema identity exceeds its byte limit"));
252    }
253    Ok(SchemaKey {
254        payload_encoding: crate::model::storage::PayloadEncoding::PostcardV1,
255        id: Identifier::new(id.as_str()).map_err(|_| invalid("invalid object schema identity"))?,
256        version: NonZeroU32::new(version).ok_or_else(|| invalid("zero object schema version"))?,
257    })
258}
259
260pub fn decode_admitted<T: StoredObject>(value: &ObjectValue, schema: &SchemaKey) -> Result<T> {
261    if &value.schema != schema {
262        return Err(invalid("object schema does not match the requested type"));
263    }
264    let decoded = T::decode(&value.payload).map_err(|error| Error::Corrupt { operation: "decode object", source: error })?;
265    decoded.validate()?;
266    let mut key = KeyMatch { expected: value.item.as_str(), offset: 0 };
267    fmt::write(&mut key, format_args!("{}", decoded.key())).map_err(|_| invalid("decoded object identity differs from stored key"))?;
268    if key.offset != value.item.as_str().len() {
269        return Err(invalid("decoded object identity differs from stored key"));
270    }
271    Ok(decoded)
272}
273
274/* === Private Functions === */
275
276fn decode<T: StoredObject>(value: &ObjectValue, max_payload_bytes: NonZeroUsize) -> Result<T> {
277    check_length(value.payload.len(), max_payload_bytes)?;
278    check_length(value.item.as_str().len(), max_payload_bytes)?;
279    decode_admitted::<T>(value, &object_schema::<T>()?)
280}
281
282fn verify<T: StoredObject>(value: &ObjectValue, max_payload_bytes: NonZeroUsize) -> Result<()> {
283    decode::<T>(value, max_payload_bytes).map(|_| ())
284}
285
286fn verify_admitted<T: StoredObject>(value: &ObjectValue, payload: NonZeroUsize, allowance: usize) -> Result<()> {
287    let overhead = std::mem::size_of::<crate::MessageError>() + MAX_IDENTITY_BYTES * 2;
288    let native = allowance.checked_sub(overhead).ok_or_else(verification_limit)?;
289    if std::mem::size_of::<T>() > native {
290        return Err(verification_limit());
291    }
292    let required = T::decoding_memory(&value.payload).map_err(|error| Error::Corrupt { operation: "preflight object", source: error })?;
293    if required > native {
294        return Err(verification_limit());
295    }
296    let decoded = decode::<T>(value, payload)?;
297    if decoded.memory_size() > required {
298        return Err(verification_limit());
299    }
300    Ok(())
301}
302
303fn verification_limit() -> Error {
304    Error::ResourceExhausted { resource: "storage object verification" }
305}
306
307/* === Tests === */
harmos_domain/model/storage/catalog.rs

harmos_domain/model/storage/
catalog.rs
