harmos_domain/state/
record.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
17
18// Third Party Imports
19use serde::{Deserialize, Serialize, de::DeserializeOwned};
20// Local Imports
21use super::envelope::{DefinitionId, FactSource, SystemFact};
22use crate::Identified;
23
24/* === Constants === */
25
26/* === Macros === */
27
28/* === Type Aliases === */
29
30pub type PreparedMembers = Vec<(Option<SystemFact>, Prepared)>;
31
32/* === Structs === */
33
34#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(deny_unknown_fields)]
36pub struct Prepared {
37    pub id: DefinitionId,
38    pub version: u32,
39    pub payload: serde_json::Value,
40}
41
42#[derive(Default)]
43pub struct Records {
44    members: Vec<(Option<SystemFact>, Prepared)>,
45    bytes: usize,
46    refusal: Option<PreparationError>,
47}
48
49/* === Enums === */
50
51#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
52pub enum PreparationError {
53    #[error("required journal capture failed")]
54    CaptureFailed,
55
56    #[error("a commit member could not be encoded")]
57    Encoding,
58
59    #[error("the commit exceeds its record or payload limit")]
60    Limit,
61
62    #[error("an emitted record is not a current member of the frozen runtime catalog")]
63    Membership,
64
65    #[error("a record has an invalid or reserved source identity")]
66    Identity,
67}
68
69/* === Traits === */
70
71pub trait Record: Identified + Serialize {}
72
73pub trait RecordProjection<R> {
74    fn project(&self) -> Option<&R>;
75}
76
77pub trait CatalogCodec {
78    fn identify(&self) -> (DefinitionId, u32);
79    fn encode<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error>;
80}
81
82pub trait RecordCatalog: CatalogCodec + Sized {
83    fn restore(id: &DefinitionId, version: u32, payload: &serde_json::Value) -> Result<Option<Self>, serde_json::Error>;
84}
85
86/* === Implementations === */
87
88impl<R> RecordProjection<R> for R {
89    fn project(&self) -> Option<&R> {
90        Some(self)
91    }
92}
93
94impl<T: Identified + Serialize> CatalogCodec for T {
95    fn identify(&self) -> (DefinitionId, u32) {
96        T::definition()
97    }
98
99    fn encode<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
100        self.serialize(serializer)
101    }
102}
103
104impl<T: Record + DeserializeOwned> RecordCatalog for T {
105    fn restore(id: &DefinitionId, version: u32, payload: &serde_json::Value) -> Result<Option<Self>, serde_json::Error> {
106        if (id, version) != (&T::definition().0, T::definition().1) {
107            return Ok(None);
108        }
109        T::deserialize(payload).map(Some)
110    }
111}
112
113impl RecordCatalog for () {
114    fn restore(_: &DefinitionId, _: u32, _: &serde_json::Value) -> Result<Option<Self>, serde_json::Error> {
115        Ok(None)
116    }
117}
118
119impl Prepared {
120    pub fn encode(value: &impl CatalogCodec) -> Result<Self, PreparationError> {
121        let (id, version) = value.identify();
122        if id.as_str().len() > Records::MAX_BYTES {
123            return Err(PreparationError::Limit);
124        }
125        struct Bounded {
126            bytes: Vec<u8>,
127            overflow: bool,
128        }
129        impl std::io::Write for Bounded {
130            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
131                if self.bytes.len().checked_add(bytes.len()).is_none_or(|n| n > Records::MAX_BYTES) {
132                    self.overflow = true;
133                    return Err(std::io::Error::other("commit payload limit"));
134                }
135                self.bytes.extend_from_slice(bytes);
136                Ok(bytes.len())
137            }
138            fn flush(&mut self) -> std::io::Result<()> {
139                Ok(())
140            }
141        }
142        let mut writer = Bounded { bytes: Vec::new(), overflow: false };
143        let result = value.encode(&mut serde_json::Serializer::new(&mut writer));
144        if writer.overflow {
145            return Err(PreparationError::Limit);
146        }
147        result.map_err(|_| PreparationError::Encoding)?;
148        let payload = serde_json::from_slice(&writer.bytes).map_err(|_| PreparationError::Encoding)?;
149        let prepared = Self { id, version, payload };
150        if prepared.bytes()? > Records::MAX_BYTES {
151            return Err(PreparationError::Limit);
152        }
153        Ok(prepared)
154    }
155
156    pub fn typed_record(value: &impl CatalogCodec) -> Result<Self, PreparationError> {
157        let prepared = Self::encode(value)?;
158        if ["harmos:work/", "harmos:guest/", "harmos:sidecar/"].iter().any(|prefix| prepared.id.as_str().starts_with(prefix)) {
159            return Err(PreparationError::Identity);
160        }
161        Ok(prepared)
162    }
163
164    pub fn system(fact: &SystemFact) -> Result<Self, PreparationError> {
165        let owner = match &fact.source {
166            FactSource::Work(owner) | FactSource::Guest(owner) | FactSource::Sidecar(owner) => owner,
167        };
168        if owner.len().checked_add(fact.name.len()).and_then(|n| n.checked_add(16)).is_none_or(|n| n > Records::MAX_BYTES) {
169            return Err(PreparationError::Limit);
170        }
171        if !fact.validate_admission() {
172            return Err(PreparationError::Identity);
173        }
174        if fact.payload.len() > Records::MAX_BYTES / 4 {
175            return Err(PreparationError::Limit);
176        }
177        let (id, version) = fact.identify();
178        let prepared = Self { id, version, payload: serde_json::json!(fact.payload) };
179        if prepared.bytes()? > Records::MAX_BYTES {
180            return Err(PreparationError::Limit);
181        }
182        Ok(prepared)
183    }
184
185    pub fn bytes(&self) -> Result<usize, PreparationError> {
186        if self.id.as_str().len() > Records::MAX_BYTES {
187            return Err(PreparationError::Limit);
188        }
189        let payload = serde_json::to_vec(&self.payload).map_err(|_| PreparationError::Encoding)?.len();
190        let identity = serde_json::to_vec(&self.id).map_err(|_| PreparationError::Encoding)?.len();
191        payload.checked_add(identity).and_then(|n| n.checked_add(4)).ok_or(PreparationError::Limit)
192    }
193}
194
195impl Records {
196    pub const MAX_RECORDS: usize = 256;
197    pub const MAX_BYTES: usize = 4 * 1024 * 1024;
198
199    pub fn push<R: Record>(&mut self, record: R) -> Result<(), PreparationError> {
200        if let Some(refusal) = self.refusal {
201            return Err(refusal);
202        }
203        let result = (|| {
204            if self.members.len() == Self::MAX_RECORDS {
205                return Err(PreparationError::Limit);
206            }
207            let prepared = Prepared::typed_record(&record)?;
208            self.reserve(prepared.bytes()?)?;
209            self.members.push((None, prepared));
210            Ok(())
211        })();
212        if let Err(refusal) = result {
213            self.refusal = Some(refusal);
214        }
215        result
216    }
217
218    pub fn push_system(&mut self, fact: SystemFact) -> Result<(), PreparationError> {
219        if let Some(refusal) = self.refusal {
220            return Err(refusal);
221        }
222        let result = (|| {
223            if self.members.len() == Self::MAX_RECORDS {
224                return Err(PreparationError::Limit);
225            }
226            let prepared = Prepared::system(&fact)?;
227            self.reserve(prepared.bytes()?)?;
228            self.members.push((Some(fact), prepared));
229            Ok(())
230        })();
231        if let Err(refusal) = result {
232            self.refusal = Some(refusal);
233        }
234        result
235    }
236
237    fn reserve(&mut self, bytes: usize) -> Result<(), PreparationError> {
238        self.bytes = self.bytes.checked_add(bytes).filter(|n| *n <= Self::MAX_BYTES).ok_or(PreparationError::Limit)?;
239        Ok(())
240    }
241
242    pub fn collect<S, T: super::Transaction<S> + CatalogCodec>(
243        transaction: &T, state: &S,
244    ) -> Result<(Prepared, PreparedMembers), PreparationError> {
245        let prepared = Prepared::encode(transaction)?;
246        let mut out = Self::default();
247        out.reserve(prepared.bytes()?)?;
248        transaction.records(state, &mut out);
249        if let Some(refusal) = out.refusal {
250            return Err(refusal);
251        }
252        Ok((prepared, out.members))
253    }
254}
255
256/* === Public Functions === */
257
258/* === Private Functions === */
259
260/* === Tests === */
harmos_domain/state/record.rs

harmos_domain/state/
record.rs
