harmos_storage/
objects.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::num::NonZeroUsize;
18// Third Party Imports
19// Local Imports
20use super::manager::{Domain, Operation};
21use super::{
22    ObjectCatalog, ObjectEntry, ObjectPoint, ObjectReceipt, ObjectRevision, ObjectValue, ObjectVersion, ObjectVersionToken, StorageManager,
23    StorageSchema, StoredObject,
24};
25use crate::Result;
26use harmos_domain::catalog::Contains;
27use harmos_domain::model::storage::catalog::{decode_admitted, object_schema};
28use harmos_domain::model::storage::object_key;
29
30/* === Constants === */
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36/* === Structs === */
37
38pub struct ObjectGroup<'a, C> {
39    manager: &'a StorageManager<C>,
40    first: u64,
41    count: usize,
42    allowance: usize,
43    used: usize,
44    schemas: Vec<StorageSchema>,
45    objects: Vec<ObjectEntry>,
46    receipts: Vec<ObjectReceipt>,
47}
48
49/* === Enums === */
50
51/* === Traits === */
52
53/* === Implementations === */
54
55impl<C: ObjectCatalog> ObjectGroup<'_, C> {
56    pub fn write<T: StoredObject>(&mut self, value: &T) -> Result<()>
57    where
58        C: Contains<T>,
59    {
60        if self.objects.len() == self.count {
61            return Err(invalid("object group contains too many objects"));
62        }
63        let settings = harmos_domain::config::SchemaSettings::default();
64        let graph =
65            crate::schema::graph::Schema::from_declaration(T::DEFINITION, &settings).map_err(|_| invalid("invalid object schema"))?;
66        let schema = object_schema::<T>()?;
67        let item = object_key(value.key(), self.manager.inner.settings.max_batch_bytes())?;
68        let payload = value.encode();
69        if payload.len() != value.encoded_size() {
70            return Err(invalid("object codec length differs"));
71        }
72        self.append(ObjectValue { schema, item, payload }, graph)
73    }
74}
75
76impl<C> ObjectGroup<'_, C> {
77    pub fn remaining_bytes(&self) -> usize {
78        self.allowance - self.used
79    }
80
81    pub(super) fn write_value(&mut self, value: ObjectValue) -> Result<()> {
82        if self.objects.len() == self.count {
83            return Err(invalid("object group contains too many objects"));
84        }
85        let definition = (self.manager.inner.object_definition)(&value.schema)?;
86        let settings = harmos_domain::config::SchemaSettings::default();
87        let graph = definition.graph(&settings)?;
88        definition.validate_admitted(&value, self.manager.inner.settings.max_batch_bytes(), settings.preparation_bytes.get())?;
89        self.append(value, graph)
90    }
91
92    fn append(&mut self, value: ObjectValue, graph: crate::schema::graph::Schema) -> Result<()> {
93        let sequence = self.first + (self.objects.len() * 2) as u64;
94        let descriptor = StorageSchema::new(self.manager.position(sequence), value.schema.clone(), graph);
95        let position = self.manager.position(sequence + 1);
96        let version = ObjectVersionToken::new(position.clone());
97        let object = ObjectEntry {
98            position: position.clone(),
99            version: version.clone(),
100            value,
101        };
102        let bytes = descriptor
103            .encoded_bytes()?
104            .checked_add(crate::manager::encoded_size(&object)?)
105            .ok_or_else(|| invalid("object size overflow"))?;
106        if bytes > self.remaining_bytes() {
107            return Err(invalid("object group exceeds its serialized byte limit"));
108        }
109        self.used += bytes;
110        self.schemas.push(descriptor);
111        self.objects.push(object);
112        self.receipts.push(ObjectReceipt { position, version });
113        Ok(())
114    }
115}
116
117impl<C: ObjectCatalog> StorageManager<C> {
118    #[tracing::instrument(name = "storage.objects.write_object", level = "debug", skip_all)]
119    pub async fn write_object<T: StoredObject>(&self, value: &T) -> Result<ObjectReceipt>
120    where
121        C: Contains<T>,
122    {
123        let receipts = self.write_object_group(self.object_admission_bytes(), NonZeroUsize::MIN, |group| group.write(value)).await?;
124        Ok(receipts.into_iter().next().unwrap())
125    }
126
127    #[tracing::instrument(name = "storage.objects.write_object_group", level = "debug", skip_all)]
128    pub async fn write_object_group(
129        &self, max_bytes: NonZeroUsize, count: NonZeroUsize, build: impl FnOnce(&mut ObjectGroup<'_, C>) -> Result<()> + Send,
130    ) -> Result<Vec<ObjectReceipt>> {
131        self.admit_object_group(max_bytes, count, None, build).await
132    }
133}
134
135impl<C: ObjectCatalog> StorageManager<C> {
136    #[tracing::instrument(name = "storage.manager.read_object", level = "debug", skip_all)]
137    pub async fn read_object<T: StoredObject>(
138        &self, key: &T::Key, version: ObjectVersion,
139    ) -> Result<Option<crate::ReadBatch<ObjectRevision<T>>>>
140    where
141        C: Contains<T>,
142    {
143        let schema = object_schema::<T>()?;
144        let budget = self.inner.settings.scan_bytes();
145        let point = ObjectPoint {
146            schema: schema.id.clone(),
147            item: object_key(key, budget)?,
148            version,
149        };
150        let saved = self.cursor_saved();
151        if let ObjectVersion::Exact(version) = &point.version {
152            version.history().check(&saved.history)?;
153            if version.revision().get() > saved.sequence {
154                return Ok(None);
155            }
156        }
157
158        let snapshot = self.snapshot(&saved).await?;
159        let Some(object) = snapshot.read_object(&point).await? else {
160            return Ok(None);
161        };
162        if matches!(&point.version, ObjectVersion::Exact(version) if version != &object.version) {
163            return Err(invalid("backend returned a different object revision"));
164        }
165        let value = decode_admitted::<T>(&object.value, &schema)?;
166        let revision = ObjectRevision {
167            value,
168            position: object.position,
169            version: object.version,
170        };
171        Ok(Some(crate::ReadBatch::new(revision)))
172    }
173}
174
175impl<C> StorageManager<C> {
176    pub(super) fn object_admission_bytes(&self) -> NonZeroUsize {
177        NonZeroUsize::new(
178            self.inner.settings.max_batch_bytes().get().saturating_mul(4).min(self.inner.settings.request.max_pending_write_bytes.get()),
179        )
180        .unwrap()
181    }
182
183    #[tracing::instrument(name = "storage.objects.admit_object_group", level = "debug", skip_all)]
184    pub(super) async fn admit_object_group(
185        &self, max_bytes: NonZeroUsize, count: NonZeroUsize, mutation: Option<super::admissions::Mutation<'_>>,
186        build: impl FnOnce(&mut ObjectGroup<'_, C>) -> Result<()> + Send,
187    ) -> Result<Vec<ObjectReceipt>> {
188        let _writer = tracing::Instrument::instrument(self.inner.object_writes.lock(), tracing::trace_span!("storage.lock.wait")).await;
189        self.admit_object_group_locked(max_bytes, count, mutation, build).await
190    }
191
192    #[tracing::instrument(name = "storage.objects.admit_object_group_locked", level = "debug", skip_all)]
193    pub(super) async fn admit_object_group_locked(
194        &self, max_bytes: NonZeroUsize, count: NonZeroUsize, mutation: Option<super::admissions::Mutation<'_>>,
195        build: impl FnOnce(&mut ObjectGroup<'_, C>) -> Result<()> + Send,
196    ) -> Result<Vec<ObjectReceipt>> {
197        let positions = count.get().checked_mul(2).ok_or_else(|| invalid("object group count overflow"))?;
198        let mut receipts = None;
199        let last = self
200            .admit_mutation(Domain::Data, max_bytes.get(), positions, mutation, false, |first| {
201                let mut group = ObjectGroup {
202                    manager: self,
203                    first,
204                    count: count.get(),
205                    allowance: max_bytes.get(),
206                    used: 0,
207                    schemas: Vec::with_capacity(count.get()),
208                    objects: Vec::with_capacity(count.get()),
209                    receipts: Vec::with_capacity(count.get()),
210                };
211                build(&mut group)?;
212                if group.objects.len() != count.get() {
213                    return Err(invalid("object group count differs from admission"));
214                }
215                receipts = Some(group.receipts);
216                Ok(Operation::ObjectGroup {
217                    schemas: super::StorageBatch::new(group.schemas),
218                    objects: super::StorageBatch::new(group.objects),
219                    bytes: group.used,
220                })
221            })
222            .await?;
223        if let Some(receipts) = receipts {
224            return Ok(receipts);
225        }
226        // A Matching Retry Reconstructs The Same Revision Tokens From Its Fixed
227        // Original Group Frontier. No Payload Is Revalidated Or Readmitted.
228        let first = last.checked_sub(positions as u64 - 2).ok_or_else(|| invalid("invalid object group receipt"))?;
229        let mut receipts = Vec::with_capacity(count.get());
230        for index in 0..count.get() {
231            let position = self.position(first + (index * 2) as u64);
232            receipts.push(ObjectReceipt {
233                version: ObjectVersionToken::new(position.clone()),
234                position,
235            });
236        }
237        Ok(receipts)
238    }
239}
240
241/* === Public Functions === */
242
243/* === Private Functions === */
244
245pub(super) fn invalid(reason: &'static str) -> crate::Error {
246    harmos_domain::model::storage::invalid(reason).into()
247}
248
249/* === Tests === */
harmos_storage/objects.rs

harmos_storage/
objects.rs
