harmos_domain/schema/
payload.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::sync::Arc;
18// Third Party Imports
19// Local Imports
20use super::{Cardinality, graph};
21use crate::MessageError;
22use crate::config::SchemaSettings;
23use crate::message::PostcardReader;
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33#[derive(Clone, Debug)]
34pub struct Validator {
35    root: usize,
36    definitions: Arc<[Box<[Checked]>]>,
37}
38#[derive(Debug)]
39struct Checked {
40    name: Box<str>,
41    tag: u32,
42    cardinality: Cardinality,
43    kind: Wire,
44}
45
46// Each Message Contributes At Most A Variant And Payload Field; Repeated
47// Indexes Consume Their Own Reader Nesting Level.
48struct Path<'schema> {
49    entries: [PathElement<'schema>; 2 * crate::message::postcard::MAX_DEPTH],
50    len: usize,
51}
52
53/* === Enums === */
54
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub enum PathElement<'schema> {
57    Field { name: &'schema str, tag: u32 },
58    Index(usize),
59}
60
61#[derive(Debug, Clone, Copy, PartialEq)]
62pub enum Scalar<'payload> {
63    Bool(bool),
64    U32(u32),
65    U64(u64),
66    I32(i32),
67    I64(i64),
68    F32(f32),
69    F64(f64),
70    String(&'payload str),
71    Bytes(&'payload [u8]),
72}
73
74#[derive(Debug, Clone, Copy, PartialEq)]
75pub enum Event<'payload> {
76    BeginMessage,
77    EndMessage,
78    Optional { present: bool },
79    BeginRepeated { length: usize },
80    EndRepeated,
81    Variant { index: u32 },
82    Scalar(Scalar<'payload>),
83}
84
85#[derive(Debug, thiserror::Error)]
86pub enum VisitError<E> {
87    #[error(transparent)]
88    Payload(#[from] PayloadError),
89
90    #[error("Postcard visitor failed: {0}")]
91    Consumer(#[source] E),
92}
93
94#[derive(Debug)]
95enum Wire {
96    Bool,
97    U32,
98    U64,
99    I32,
100    I64,
101    F32,
102    F64,
103    String,
104    Bytes,
105    Message(usize),
106}
107#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
108#[error(transparent)]
109pub struct PayloadError(#[from] MessageError);
110
111/* === Traits === */
112
113/* === Implementations === */
114
115impl<'schema> Path<'schema> {
116    fn new() -> Self {
117        Self {
118            entries: [PathElement::Index(0); 2 * crate::message::postcard::MAX_DEPTH],
119            len: 0,
120        }
121    }
122
123    fn push(&mut self, element: PathElement<'schema>) {
124        self.entries[self.len] = element;
125        self.len += 1;
126    }
127
128    fn pop(&mut self) {
129        self.len -= 1;
130    }
131}
132
133impl<'schema> std::ops::Deref for Path<'schema> {
134    type Target = [PathElement<'schema>];
135
136    fn deref(&self) -> &Self::Target {
137        &self.entries[..self.len]
138    }
139}
140
141impl Validator {
142    // Reuse The Checked Definitions For Another Root Without Recompiling Them.
143    pub fn at(&self, root: usize) -> Result<Self, graph::Error> {
144        if root >= self.definitions.len() {
145            return Err(graph::Error::Reference { definition: root });
146        }
147        Ok(Self {
148            root,
149            definitions: Arc::clone(&self.definitions),
150        })
151    }
152    pub fn construction_memory_bytes(schema: &graph::Schema, settings: &SchemaSettings) -> Result<usize, graph::Error> {
153        schema
154            .preparation(settings)?
155            .bytes
156            .checked_add(graph_compilation_bytes(schema)?)
157            .and_then(|bytes| bytes.checked_add(2 * std::mem::size_of::<usize>()))
158            .and_then(|bytes| {
159                schema
160                    .definitions
161                    .len()
162                    .checked_mul(std::mem::size_of::<Box<[Checked]>>())
163                    .and_then(|temporary| bytes.checked_add(temporary))
164            })
165            .ok_or(graph::Error::Budget)
166    }
167    pub fn validate(&self, bytes: &[u8]) -> Result<(), PayloadError> {
168        match self.visit(bytes, |_, _| Ok::<(), std::convert::Infallible>(())) {
169            Ok(()) => Ok(()),
170            Err(VisitError::Payload(error)) => Err(error),
171            Err(VisitError::Consumer(error)) => match error {},
172        }
173    }
174
175    pub fn visit<'schema, 'payload, E>(
176        &'schema self, bytes: &'payload [u8], mut visitor: impl FnMut(&[PathElement<'schema>], Event<'payload>) -> Result<(), E>,
177    ) -> Result<(), VisitError<E>> {
178        let mut input = PostcardReader::new(bytes);
179        let mut path = Path::new();
180        let mut failure = None;
181        let result = self.read(self.root, &mut input, &mut path, &mut |path, event| {
182            visitor(path, event).map_err(|error| {
183                failure = Some(error);
184                MessageError::borrowed("Postcard consumer failed")
185            })
186        });
187        if let Some(error) = failure {
188            return Err(VisitError::Consumer(error));
189        }
190        result.map_err(PayloadError::from)?;
191        input.finish().map_err(PayloadError::from)?;
192        Ok(())
193    }
194    pub fn from_schema(schema: &graph::Schema, settings: &SchemaSettings) -> Result<Self, graph::Error> {
195        schema.prepare(settings)?;
196        if Self::construction_memory_bytes(schema, settings)? > settings.preparation_bytes.get() {
197            return Err(graph::Error::Budget);
198        }
199        let definitions = schema
200            .definitions
201            .iter()
202            .map(|fields| {
203                fields
204                    .iter()
205                    .map(|field| Checked {
206                        name: field.name.as_ref().into(),
207                        tag: field.tag,
208                        cardinality: field.cardinality,
209                        kind: match field.kind {
210                            graph::Kind::Bool => Wire::Bool,
211                            graph::Kind::U32 => Wire::U32,
212                            graph::Kind::U64 => Wire::U64,
213                            graph::Kind::I32 => Wire::I32,
214                            graph::Kind::I64 => Wire::I64,
215                            graph::Kind::F32 => Wire::F32,
216                            graph::Kind::F64 => Wire::F64,
217                            graph::Kind::String => Wire::String,
218                            graph::Kind::Bytes => Wire::Bytes,
219                            graph::Kind::Message { definition } => Wire::Message(definition),
220                        },
221                    })
222                    .collect::<Box<[_]>>()
223            })
224            .collect::<Box<[_]>>();
225        Ok(Self {
226            root: schema.root,
227            definitions: definitions.into(),
228        })
229    }
230    fn read<'schema, 'payload>(
231        &'schema self, definition: usize, input: &mut PostcardReader<'payload>, path: &mut Path<'schema>,
232        visitor: &mut impl FnMut(&[PathElement<'schema>], Event<'payload>) -> Result<(), MessageError>,
233    ) -> Result<(), MessageError> {
234        input.nested(|input| {
235            visitor(path, Event::BeginMessage)?;
236            if self.definitions[definition].first().is_some_and(|field| field.cardinality == Cardinality::Variant) {
237                let index: u32 = input.value()?;
238                let field = self.definitions[definition]
239                    .get(index as usize)
240                    .ok_or_else(|| MessageError::borrowed("unknown Postcard enum variant"))?;
241                let Wire::Message(payload) = &field.kind else { unreachable!("validated enum schema") };
242                path.push(PathElement::Field { name: &field.name, tag: field.tag });
243                visitor(path, Event::Variant { index })?;
244                visitor(path, Event::BeginMessage)?;
245                self.read_fields(*payload, input, path, visitor)?;
246                visitor(path, Event::EndMessage)?;
247                path.pop();
248            } else {
249                self.read_fields(definition, input, path, visitor)?;
250            }
251            visitor(path, Event::EndMessage)
252        })
253    }
254    fn read_fields<'schema, 'payload>(
255        &'schema self, definition: usize, input: &mut PostcardReader<'payload>, path: &mut Path<'schema>,
256        visitor: &mut impl FnMut(&[PathElement<'schema>], Event<'payload>) -> Result<(), MessageError>,
257    ) -> Result<(), MessageError> {
258        for field in &self.definitions[definition] {
259            path.push(PathElement::Field { name: &field.name, tag: field.tag });
260            match field.cardinality {
261                Cardinality::Required | Cardinality::Variant => field.kind.read(self, input, path, visitor)?,
262                Cardinality::Optional => input.nested(|input| {
263                    let present = input.value::<bool>()?;
264                    visitor(path, Event::Optional { present })?;
265                    if present {
266                        field.kind.read(self, input, path, visitor)?;
267                    }
268                    Ok(())
269                })?,
270                Cardinality::Repeated => input.nested(|input| {
271                    let count: usize = input.value()?;
272                    input.claim(count)?;
273                    visitor(path, Event::BeginRepeated { length: count })?;
274                    for index in 0..count {
275                        path.push(PathElement::Index(index));
276                        field.kind.read(self, input, path, visitor)?;
277                        path.pop();
278                    }
279                    visitor(path, Event::EndRepeated)
280                })?,
281            }
282            path.pop();
283        }
284        Ok(())
285    }
286}
287impl Wire {
288    fn read<'schema, 'payload>(
289        &'schema self, schema: &'schema Validator, input: &mut PostcardReader<'payload>, path: &mut Path<'schema>,
290        visitor: &mut impl FnMut(&[PathElement<'schema>], Event<'payload>) -> Result<(), MessageError>,
291    ) -> Result<(), MessageError> {
292        let scalar = match self {
293            Self::Bool => Scalar::Bool(input.value()?),
294            Self::U32 => Scalar::U32(input.value()?),
295            Self::U64 => Scalar::U64(input.value()?),
296            Self::I32 => Scalar::I32(input.value()?),
297            Self::I64 => Scalar::I64(input.value()?),
298            Self::F32 => Scalar::F32(input.value()?),
299            Self::F64 => Scalar::F64(input.value()?),
300            Self::String => Scalar::String(input.value()?),
301            Self::Bytes => Scalar::Bytes(input.nested(|input| {
302                let count = input.value()?;
303                input.take(count)
304            })?),
305            Self::Message(definition) => return schema.read(*definition, input, path, visitor),
306        };
307        visitor(path, Event::Scalar(scalar))
308    }
309}
310impl crate::MemorySize for Validator {
311    fn memory_size(&self) -> usize {
312        let mut bytes = std::mem::size_of::<Self>()
313            .saturating_add(2 * std::mem::size_of::<usize>())
314            .saturating_add(self.definitions.len().saturating_mul(std::mem::size_of::<Box<[Checked]>>()));
315        for fields in self.definitions.iter() {
316            bytes = bytes.saturating_add(fields.len().saturating_mul(std::mem::size_of::<Checked>()));
317            for field in fields {
318                bytes = bytes.saturating_add(field.name.len());
319            }
320        }
321        bytes
322    }
323}
324
325/* === Public Functions === */
326
327/* === Private Functions === */
328
329fn graph_compilation_bytes(schema: &graph::Schema) -> Result<usize, graph::Error> {
330    let mut bytes = std::mem::size_of::<Validator>()
331        .checked_add(schema.definitions.len().checked_mul(std::mem::size_of::<Box<[Checked]>>()).ok_or(graph::Error::Budget)?)
332        .ok_or(graph::Error::Budget)?;
333    for fields in &schema.definitions {
334        bytes = bytes
335            .checked_add(fields.len().checked_mul(std::mem::size_of::<Checked>()).ok_or(graph::Error::Budget)?)
336            .ok_or(graph::Error::Budget)?;
337        for field in fields {
338            bytes = bytes.checked_add(field.name.len()).ok_or(graph::Error::Budget)?;
339        }
340    }
341    Ok(bytes)
342}
343
344/* === Tests === */
345
346#[cfg(test)]
347mod tests {
348    use super::*;
349
350    fn enum_chain(depth: usize) -> (Validator, Vec<u8>) {
351        let mut definitions = Vec::new();
352        for level in 0..depth {
353            let variant = Checked {
354                name: "variant".into(),
355                tag: 1,
356                cardinality: Cardinality::Variant,
357                kind: Wire::Message(2 * level + 1),
358            };
359            let kind = if level + 1 == depth { Wire::Bool } else { Wire::Message(2 * level + 2) };
360            let field = Checked {
361                name: "value".into(),
362                tag: 1,
363                cardinality: Cardinality::Required,
364                kind,
365            };
366            definitions.push(vec![variant].into_boxed_slice());
367            definitions.push(vec![field].into_boxed_slice());
368        }
369        let validator = Validator { root: 0, definitions: definitions.into() };
370        let mut bytes = vec![0; depth];
371        bytes.push(1);
372        (validator, bytes)
373    }
374
375    #[test]
376    fn deepest_enum_path_preserves_events_and_consumer_errors() {
377        let depth = crate::message::postcard::MAX_DEPTH;
378        let (validator, bytes) = enum_chain(depth);
379        let mut longest = 0;
380        validator
381            .visit(&bytes, |path, event| {
382                longest = longest.max(path.len());
383                if let Event::Scalar(value) = event {
384                    assert_eq!(value, Scalar::Bool(true));
385                    assert_eq!(path.len(), 2 * depth);
386                }
387                Ok::<_, ()>(())
388            })
389            .unwrap();
390        assert_eq!(longest, 2 * depth);
391        let failure = validator.visit(&bytes, |_, event| if matches!(event, Event::Scalar(_)) { Err(7) } else { Ok(()) });
392        assert!(matches!(failure, Err(VisitError::Consumer(7))));
393    }
394
395    #[test]
396    fn excessive_enum_depth_returns_payload_error() {
397        let (validator, bytes) = enum_chain(crate::message::postcard::MAX_DEPTH + 1);
398        let result = validator.visit(&bytes, |_, _| Ok::<_, ()>(()));
399        assert!(matches!(result, Err(VisitError::Payload(_))));
400    }
401
402    #[test]
403    fn repeated_index_uses_its_existing_nesting_level() {
404        let depth = crate::message::postcard::MAX_DEPTH - 1;
405        let (mut validator, mut bytes) = enum_chain(depth);
406        let definitions = Arc::get_mut(&mut validator.definitions).unwrap();
407        definitions.last_mut().unwrap()[0].cardinality = Cardinality::Repeated;
408        bytes.insert(depth, 1);
409        let mut scalars = 0;
410        validator
411            .visit(&bytes, |path, event| {
412                if matches!(event, Event::Scalar(Scalar::Bool(true))) {
413                    assert_eq!(path.len(), 2 * depth + 1);
414                    assert_eq!(path.last(), Some(&PathElement::Index(0)));
415                    scalars += 1;
416                }
417                Ok::<_, ()>(())
418            })
419            .unwrap();
420        assert_eq!(scalars, 1);
421    }
422}
harmos_domain/schema/payload.rs

harmos_domain/schema/
payload.rs
