harmos_domain/schema/
graph.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Shared Message Schema Definitions And References
11
12/* === Modules === */
13
14mod codec;
15
16/* === Re-Exports === */
17
18/* === Imports === */
19
20// Standard Library Imports
21use std::borrow::Cow;
22use std::collections::HashSet;
23use std::fmt;
24use std::mem::size_of;
25// Third Party Imports
26use serde::de::{DeserializeSeed, Error as _, MapAccess, SeqAccess, Visitor};
27use serde::{Deserialize, Deserializer, Serialize};
28// Local Imports
29use super::{Cardinality, MAX_NAME_BYTES, diagnostic_name};
30use crate::config::SchemaSettings;
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38/* === Structs === */
39
40#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
41pub struct Schema {
42    pub root: usize,
43    pub definitions: Vec<Vec<Field>>,
44}
45
46#[derive(Clone, Copy, Debug)]
47pub struct View<'a> {
48    schema: &'a Schema,
49    root: usize,
50}
51
52#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
53#[serde(deny_unknown_fields)]
54pub struct Field<R = usize> {
55    #[serde(deserialize_with = "name")]
56    pub name: Cow<'static, str>,
57
58    pub tag: u32,
59    pub kind: Kind<R>,
60    pub cardinality: Cardinality,
61}
62
63#[derive(Clone, Copy, Debug, PartialEq, Eq)]
64pub struct Preparation {
65    pub definitions: usize,
66    pub fields: usize,
67    pub bytes: usize,
68    pub work: usize,
69}
70
71pub struct Seed<'a> {
72    pub settings: &'a SchemaSettings,
73}
74
75struct DefinitionsSeed<'a>(&'a mut Budget);
76struct DefinitionSeed<'a>(&'a mut Budget);
77struct FieldSeed<'a>(&'a mut Budget);
78struct Budget {
79    remaining: usize,
80}
81
82/* === Enums === */
83
84#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize)]
85#[serde(tag = "type", rename_all = "snake_case")]
86pub enum Kind<R = usize> {
87    Bool,
88    U32,
89    U64,
90    I32,
91    I64,
92    F32,
93    F64,
94    String,
95    Bytes,
96    Message { definition: R },
97}
98
99#[derive(Debug, thiserror::Error)]
100pub enum Error {
101    #[error("schema preparation exceeds its memory or work allowance")]
102    Budget,
103
104    #[error("schema definition {definition} is absent")]
105    Reference { definition: usize },
106
107    #[error("schema definition {definition} is unreachable")]
108    Unreachable { definition: usize },
109
110    #[error("invalid schema field `{field}`: {reason}")]
111    Field { field: String, reason: &'static str },
112}
113
114/* === Traits === */
115
116/* === Implementations === */
117
118impl<R> Kind<R> {
119    pub(crate) fn try_map<S, E>(self, reference: impl FnOnce(R) -> Result<S, E>) -> Result<Kind<S>, E> {
120        Ok(match self {
121            Self::Bool => Kind::Bool,
122            Self::U32 => Kind::U32,
123            Self::U64 => Kind::U64,
124            Self::I32 => Kind::I32,
125            Self::I64 => Kind::I64,
126            Self::F32 => Kind::F32,
127            Self::F64 => Kind::F64,
128            Self::String => Kind::String,
129            Self::Bytes => Kind::Bytes,
130            Self::Message { definition } => Kind::Message { definition: reference(definition)? },
131        })
132    }
133}
134
135impl<'de, R: Deserialize<'de>> Deserialize<'de> for Kind<R> {
136    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
137        #[derive(Deserialize)]
138        #[serde(deny_unknown_fields, bound(deserialize = "R: Deserialize<'de>"))]
139        struct Input<R> {
140            #[serde(rename = "type", deserialize_with = "kind_name")]
141            kind: String,
142
143            #[serde(default, deserialize_with = "definition")]
144            definition: Option<R>,
145        }
146        let input = Input::<R>::deserialize(deserializer)?;
147        if input.kind == "message" {
148            return input.definition.map(|definition| Self::Message { definition }).ok_or_else(|| D::Error::missing_field("definition"));
149        }
150        if input.definition.is_some() {
151            return Err(D::Error::custom("scalar schema kind cannot reference a definition"));
152        }
153        match input.kind.as_str() {
154            "bool" => Ok(Self::Bool),
155            "u32" => Ok(Self::U32),
156            "u64" => Ok(Self::U64),
157            "i32" => Ok(Self::I32),
158            "i64" => Ok(Self::I64),
159            "f32" => Ok(Self::F32),
160            "f64" => Ok(Self::F64),
161            "string" => Ok(Self::String),
162            "bytes" => Ok(Self::Bytes),
163            _ => Err(D::Error::custom("unknown schema kind")),
164        }
165    }
166}
167
168impl crate::MemorySize for Schema {
169    fn memory_size(&self) -> usize {
170        let mut bytes = size_of::<Self>().saturating_add(self.definitions.capacity().saturating_mul(size_of::<Vec<Field>>()));
171        for fields in &self.definitions {
172            bytes = bytes.saturating_add(fields.capacity().saturating_mul(size_of::<Field>()));
173            for field in fields {
174                if let Cow::Owned(name) = &field.name {
175                    bytes = bytes.saturating_add(name.capacity());
176                }
177            }
178        }
179        bytes
180    }
181}
182
183impl Schema {
184    pub fn clone_memory_bytes(&self) -> Result<usize, Error> {
185        let mut bytes = self
186            .definitions
187            .len()
188            .checked_mul(size_of::<Vec<Field>>())
189            .and_then(|bytes| bytes.checked_add(size_of::<Self>()))
190            .ok_or(Error::Budget)?;
191        for fields in &self.definitions {
192            bytes = fields.len().checked_mul(size_of::<Field>()).and_then(|fields| bytes.checked_add(fields)).ok_or(Error::Budget)?;
193            for field in fields {
194                if let Cow::Owned(name) = &field.name {
195                    bytes = bytes.checked_add(name.len()).ok_or(Error::Budget)?;
196                }
197            }
198        }
199        Ok(bytes)
200    }
201
202    pub fn validation_memory_bytes(&self, settings: &SchemaSettings) -> Result<usize, Error> {
203        let preparation = self.preparation(settings)?;
204        preparation.bytes.checked_sub(crate::MemorySize::memory_size(self)).ok_or(Error::Budget)
205    }
206
207    pub fn view(&self, root: usize) -> Result<View<'_>, Error> {
208        if root >= self.definitions.len() {
209            return Err(Error::Reference { definition: root });
210        }
211        Ok(View { schema: self, root })
212    }
213
214    pub fn root(&self) -> Result<View<'_>, Error> {
215        self.view(self.root)
216    }
217
218    // Compare Wire Shapes Independently Of Local Definition Indices.
219    pub fn equivalent(&self, root: usize, other: &Self, other_root: usize, settings: &SchemaSettings) -> Result<bool, Error> {
220        let mut preparation = self.prepare(settings)?;
221        let right = other.prepare(settings)?;
222        if root >= self.definitions.len() {
223            return Err(Error::Reference { definition: root });
224        }
225        if other_root >= other.definitions.len() {
226            return Err(Error::Reference { definition: other_root });
227        }
228        if !std::ptr::eq(self, other) {
229            preparation.add_bytes(right.bytes, 1, settings)?;
230            preparation.add_work(right.work, settings)?;
231        }
232        preparation.add_bytes(4, size_of::<(usize, usize)>(), settings)?;
233        let mut pending = vec![(root, other_root)];
234        let mut visited = HashSet::new();
235        while let Some((left, right)) = pending.pop() {
236            preparation.add_work(1, settings)?;
237            if visited.contains(&(left, right)) {
238                continue;
239            }
240            preparation.add_bytes(4, size_of::<(usize, usize)>(), settings)?;
241            visited.insert((left, right));
242            let left = &self.definitions[left];
243            let right = &other.definitions[right];
244            if left.len() != right.len() {
245                return Ok(false);
246            }
247            for (left, right) in left.iter().zip(right) {
248                preparation.add_work(left.name.len() + 1, settings)?;
249                if left.name != right.name || left.tag != right.tag || left.cardinality != right.cardinality {
250                    return Ok(false);
251                }
252                match (left.kind, right.kind) {
253                    (Kind::Message { definition: left }, Kind::Message { definition: right }) => {
254                        preparation.add_bytes(4, size_of::<(usize, usize)>(), settings)?;
255                        pending.push((left, right));
256                    }
257                    (left, right) if left == right => {}
258                    _ => return Ok(false),
259                }
260            }
261        }
262        Ok(true)
263    }
264
265    pub fn preparation(&self, settings: &SchemaSettings) -> Result<Preparation, Error> {
266        if self.root >= self.definitions.len() {
267            return Err(Error::Reference { definition: self.root });
268        }
269        let mut preparation = Preparation {
270            definitions: self.definitions.len(),
271            fields: 0,
272            bytes: size_of::<Self>(),
273            work: 0,
274        };
275        // Charge Retained Storage And Traversal Scratch Before Allocating Scratch.
276        preparation.add_bytes(self.definitions.capacity(), size_of::<Vec<Field>>(), settings)?;
277        preparation.add_bytes(self.definitions.len(), size_of::<bool>() + size_of::<usize>(), settings)?;
278        for fields in &self.definitions {
279            preparation.fields = preparation.fields.checked_add(fields.len()).ok_or(Error::Budget)?;
280            preparation.add_bytes(fields.capacity(), size_of::<Field>(), settings)?;
281            preparation.add_work(fields.len() + 1, settings)?;
282            for field in fields {
283                if let Cow::Owned(name) = &field.name {
284                    preparation.add_bytes(name.capacity(), 1, settings)?;
285                }
286                preparation.add_work(field.name.len(), settings)?;
287            }
288        }
289        let scratch = self.definitions.iter().map(|fields| fields.len()).max().unwrap_or(0);
290        preparation.add_bytes(scratch, size_of::<&str>() + size_of::<u32>(), settings)?;
291        for fields in &self.definitions {
292            let bits = usize::BITS as usize - fields.len().leading_zeros() as usize;
293            let work = fields.len().checked_mul(2 * bits + 2).ok_or(Error::Budget)?;
294            preparation.add_work(work, settings)?;
295        }
296        preparation.add_work(self.definitions.len(), settings)?;
297        preparation.add_work(preparation.fields, settings)?;
298        Ok(preparation)
299    }
300
301    pub fn prepare(&self, settings: &SchemaSettings) -> Result<Preparation, Error> {
302        let preparation = self.preparation(settings)?;
303        for fields in &self.definitions {
304            // Sorting Is Bounded By Direct Fields, Independent Of Repeated References.
305            let mut names = Vec::with_capacity(fields.len());
306            let mut tags = Vec::with_capacity(fields.len());
307            let variants = fields.iter().any(|field| field.cardinality == Cardinality::Variant);
308            for field in fields {
309                let invalid = |reason| Error::Field {
310                    field: diagnostic_name(&field.name),
311                    reason,
312                };
313                if variants && (field.cardinality != Cardinality::Variant || !matches!(field.kind, Kind::Message { .. })) {
314                    return Err(invalid("an enum schema requires only message-valued variants"));
315                }
316                if field.name.is_empty() || field.name.len() > MAX_NAME_BYTES || field.name.chars().any(char::is_control) {
317                    return Err(invalid("name must be nonempty, at most 256 bytes, and contain no control characters"));
318                }
319                if field.tag == 0 || field.tag > 536_870_911 || (19_000..=19_999).contains(&field.tag) {
320                    return Err(invalid("invalid protobuf tag"));
321                }
322                if let Kind::Message { definition } = field.kind
323                    && definition >= self.definitions.len()
324                {
325                    return Err(Error::Reference { definition });
326                }
327                names.push(field.name.as_ref());
328                tags.push(field.tag);
329            }
330            names.sort_unstable();
331            tags.sort_unstable();
332            if let Some(pair) = names.windows(2).find(|pair| pair[0] == pair[1]) {
333                return Err(Error::Field {
334                    field: diagnostic_name(pair[0]),
335                    reason: "duplicate name",
336                });
337            }
338            if let Some(pair) = tags.windows(2).find(|pair| pair[0] == pair[1]) {
339                let field = fields.iter().find(|field| field.tag == pair[0]).unwrap();
340                return Err(Error::Field {
341                    field: diagnostic_name(&field.name),
342                    reason: "duplicate protobuf tag",
343                });
344            }
345        }
346        let mut reached = vec![false; self.definitions.len()];
347        let mut pending = Vec::with_capacity(self.definitions.len());
348        reached[self.root] = true;
349        pending.push(self.root);
350        while let Some(definition) = pending.pop() {
351            for field in &self.definitions[definition] {
352                if let Kind::Message { definition } = field.kind
353                    && !reached[definition]
354                {
355                    reached[definition] = true;
356                    pending.push(definition);
357                }
358            }
359        }
360        if let Some(definition) = reached.iter().position(|reached| !reached) {
361            return Err(Error::Unreachable { definition });
362        }
363        Ok(preparation)
364    }
365}
366
367impl<'a> View<'a> {
368    pub fn schema(self) -> &'a Schema {
369        self.schema
370    }
371    pub fn index(self) -> usize {
372        self.root
373    }
374    pub fn fields(self) -> &'a [Field] {
375        &self.schema.definitions[self.root]
376    }
377
378    pub fn message(self, field: &Field) -> Result<Option<Self>, Error> {
379        match field.kind {
380            Kind::Message { definition } => self.schema.view(definition).map(Some),
381            _ => Ok(None),
382        }
383    }
384}
385
386impl Preparation {
387    fn add_bytes(&mut self, count: usize, size: usize, settings: &SchemaSettings) -> Result<(), Error> {
388        self.bytes = count.checked_mul(size).and_then(|bytes| self.bytes.checked_add(bytes)).ok_or(Error::Budget)?;
389        if self.bytes > settings.preparation_bytes.get() {
390            return Err(Error::Budget);
391        }
392        Ok(())
393    }
394
395    fn add_work(&mut self, work: usize, settings: &SchemaSettings) -> Result<(), Error> {
396        self.work = self.work.checked_add(work).ok_or(Error::Budget)?;
397        if self.work > settings.preparation_work.get() {
398            return Err(Error::Budget);
399        }
400        Ok(())
401    }
402}
403
404impl<'de> Deserialize<'de> for Schema {
405    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
406        Seed { settings: &SchemaSettings::default() }.deserialize(deserializer)
407    }
408}
409
410impl<'de> DeserializeSeed<'de> for Seed<'_> {
411    type Value = Schema;
412    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Schema, D::Error> {
413        struct Input<'a>(&'a SchemaSettings);
414        impl<'de> Visitor<'de> for Input<'_> {
415            type Value = Schema;
416            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
417                formatter.write_str("a bounded shared message schema")
418            }
419            fn visit_map<A: MapAccess<'de>>(self, mut input: A) -> Result<Schema, A::Error> {
420                let mut root = None;
421                let mut definitions = None;
422                let mut budget = Budget {
423                    remaining: self.0.preparation_bytes.get(),
424                };
425                while let Some(key) = input.next_key::<String>()? {
426                    match key.as_str() {
427                        "root" => {
428                            if root.is_some() {
429                                return Err(A::Error::duplicate_field("root"));
430                            }
431                            root = Some(input.next_value()?);
432                        }
433                        "definitions" => {
434                            if definitions.is_some() {
435                                return Err(A::Error::duplicate_field("definitions"));
436                            }
437                            definitions = Some(input.next_value_seed(DefinitionsSeed(&mut budget))?);
438                        }
439                        _ => return Err(A::Error::unknown_field(&key, &["root", "definitions"])),
440                    }
441                }
442                let schema = Schema {
443                    root: root.ok_or_else(|| A::Error::missing_field("root"))?,
444                    definitions: definitions.ok_or_else(|| A::Error::missing_field("definitions"))?,
445                };
446                schema.prepare(self.0).map_err(A::Error::custom)?;
447                Ok(schema)
448            }
449        }
450        deserializer.deserialize_map(Input(self.settings))
451    }
452}
453
454impl Budget {
455    fn charge<E: serde::de::Error>(&mut self, bytes: usize) -> Result<(), E> {
456        self.remaining = self.remaining.checked_sub(bytes).ok_or_else(|| E::custom(Error::Budget))?;
457        Ok(())
458    }
459}
460
461impl<'de> DeserializeSeed<'de> for DefinitionsSeed<'_> {
462    type Value = Vec<Vec<Field>>;
463    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
464        deserializer.deserialize_seq(self)
465    }
466}
467
468impl<'de> DeserializeSeed<'de> for DefinitionSeed<'_> {
469    type Value = Vec<Field>;
470    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
471        deserializer.deserialize_seq(self)
472    }
473}
474
475impl<'de> DeserializeSeed<'de> for FieldSeed<'_> {
476    type Value = Field;
477    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
478        // Reserve Capacity And Sorting Scratch Before Any Field Allocation.
479        self.0.charge::<D::Error>(4 * size_of::<Field>() + size_of::<&str>() + size_of::<u32>() + MAX_NAME_BYTES)?;
480        Field::deserialize(deserializer)
481    }
482}
483
484impl<'de> Visitor<'de> for DefinitionsSeed<'_> {
485    type Value = Vec<Vec<Field>>;
486    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
487        formatter.write_str("bounded message definitions")
488    }
489    fn visit_seq<A: SeqAccess<'de>>(self, mut input: A) -> Result<Self::Value, A::Error> {
490        let mut definitions = Vec::new();
491        while let Some(fields) = input.next_element_seed(DefinitionSeed(self.0))? {
492            definitions.push(fields);
493        }
494        Ok(definitions)
495    }
496}
497
498impl<'de> Visitor<'de> for DefinitionSeed<'_> {
499    type Value = Vec<Field>;
500    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
501        formatter.write_str("bounded direct message fields")
502    }
503    fn visit_seq<A: SeqAccess<'de>>(self, mut input: A) -> Result<Self::Value, A::Error> {
504        self.0.charge::<A::Error>(4 * size_of::<Vec<Field>>() + size_of::<usize>() + size_of::<bool>())?;
505        let mut fields = Vec::new();
506        while let Some(field) = input.next_element_seed(FieldSeed(self.0))? {
507            fields.push(field);
508        }
509        Ok(fields)
510    }
511}
512
513/* === Public Functions === */
514
515/* === Private Functions === */
516
517fn definition<'de, D: Deserializer<'de>, R: Deserialize<'de>>(deserializer: D) -> Result<Option<R>, D::Error> {
518    R::deserialize(deserializer).map(Some)
519}
520
521fn kind_name<'de, D: Deserializer<'de>>(deserializer: D) -> Result<String, D::Error> {
522    name(deserializer).map(Cow::into_owned)
523}
524
525fn name<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Cow<'static, str>, D::Error> {
526    struct Name;
527    impl<'de> Visitor<'de> for Name {
528        type Value = String;
529        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
530            formatter.write_str("a field name of at most 256 bytes")
531        }
532        fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<String, E> {
533            if value.len() > MAX_NAME_BYTES {
534                return Err(E::custom("schema field name exceeds 256 bytes"));
535            }
536            Ok(value.to_owned())
537        }
538        fn visit_string<E: serde::de::Error>(self, value: String) -> Result<String, E> {
539            if value.len() > MAX_NAME_BYTES {
540                return Err(E::custom("schema field name exceeds 256 bytes"));
541            }
542            Ok(value)
543        }
544    }
545    deserializer.deserialize_string(Name).map(Cow::Owned)
546}
547
548/* === Tests === */
harmos_domain/schema/graph.rs

harmos_domain/schema/
graph.rs
