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 { definitions: self.definitions.len(), fields: 0, bytes: size_of::<Self>(), work: 0 };
270        // Charge Retained Storage And Traversal Scratch Before Allocating Scratch.
271        preparation.add_bytes(self.definitions.capacity(), size_of::<Vec<Field>>(), settings)?;
272        preparation.add_bytes(self.definitions.len(), size_of::<bool>() + size_of::<usize>(), settings)?;
273        for fields in &self.definitions {
274            preparation.fields = preparation.fields.checked_add(fields.len()).ok_or(Error::Budget)?;
275            preparation.add_bytes(fields.capacity(), size_of::<Field>(), settings)?;
276            preparation.add_work(fields.len() + 1, settings)?;
277            for field in fields {
278                if let Cow::Owned(name) = &field.name {
279                    preparation.add_bytes(name.capacity(), 1, settings)?;
280                }
281                preparation.add_work(field.name.len(), settings)?;
282            }
283        }
284        let scratch = self.definitions.iter().map(|fields| fields.len()).max().unwrap_or(0);
285        preparation.add_bytes(scratch, size_of::<&str>() + size_of::<u32>(), settings)?;
286        for fields in &self.definitions {
287            let bits = usize::BITS as usize - fields.len().leading_zeros() as usize;
288            let work = fields.len().checked_mul(2 * bits + 2).ok_or(Error::Budget)?;
289            preparation.add_work(work, settings)?;
290        }
291        preparation.add_work(self.definitions.len(), settings)?;
292        preparation.add_work(preparation.fields, settings)?;
293        Ok(preparation)
294    }
295
296    pub fn prepare(&self, settings: &SchemaSettings) -> Result<Preparation, Error> {
297        let preparation = self.preparation(settings)?;
298        for fields in &self.definitions {
299            // Sorting Is Bounded By Direct Fields, Independent Of Repeated References.
300            let mut names = Vec::with_capacity(fields.len());
301            let mut tags = Vec::with_capacity(fields.len());
302            let variants = fields.iter().any(|field| field.cardinality == Cardinality::Variant);
303            for field in fields {
304                let invalid = |reason| Error::Field { field: diagnostic_name(&field.name), reason };
305                if variants && (field.cardinality != Cardinality::Variant || !matches!(field.kind, Kind::Message { .. })) {
306                    return Err(invalid("an enum schema requires only message-valued variants"));
307                }
308                if field.name.is_empty() || field.name.len() > MAX_NAME_BYTES || field.name.chars().any(char::is_control) {
309                    return Err(invalid("name must be nonempty, at most 256 bytes, and contain no control characters"));
310                }
311                if field.tag == 0 || field.tag > 536_870_911 || (19_000..=19_999).contains(&field.tag) {
312                    return Err(invalid("invalid protobuf tag"));
313                }
314                if let Kind::Message { definition } = field.kind
315                    && definition >= self.definitions.len()
316                {
317                    return Err(Error::Reference { definition });
318                }
319                names.push(field.name.as_ref());
320                tags.push(field.tag);
321            }
322            names.sort_unstable();
323            tags.sort_unstable();
324            if let Some(pair) = names.windows(2).find(|pair| pair[0] == pair[1]) {
325                return Err(Error::Field { field: diagnostic_name(pair[0]), reason: "duplicate name" });
326            }
327            if let Some(pair) = tags.windows(2).find(|pair| pair[0] == pair[1]) {
328                let field = fields.iter().find(|field| field.tag == pair[0]).unwrap();
329                return Err(Error::Field { field: diagnostic_name(&field.name), reason: "duplicate protobuf tag" });
330            }
331        }
332        let mut reached = vec![false; self.definitions.len()];
333        let mut pending = Vec::with_capacity(self.definitions.len());
334        reached[self.root] = true;
335        pending.push(self.root);
336        while let Some(definition) = pending.pop() {
337            for field in &self.definitions[definition] {
338                if let Kind::Message { definition } = field.kind
339                    && !reached[definition]
340                {
341                    reached[definition] = true;
342                    pending.push(definition);
343                }
344            }
345        }
346        if let Some(definition) = reached.iter().position(|reached| !reached) {
347            return Err(Error::Unreachable { definition });
348        }
349        Ok(preparation)
350    }
351}
352
353impl<'a> View<'a> {
354    pub fn schema(self) -> &'a Schema {
355        self.schema
356    }
357    pub fn index(self) -> usize {
358        self.root
359    }
360    pub fn fields(self) -> &'a [Field] {
361        &self.schema.definitions[self.root]
362    }
363
364    pub fn message(self, field: &Field) -> Result<Option<Self>, Error> {
365        match field.kind {
366            Kind::Message { definition } => self.schema.view(definition).map(Some),
367            _ => Ok(None),
368        }
369    }
370}
371
372impl Preparation {
373    fn add_bytes(&mut self, count: usize, size: usize, settings: &SchemaSettings) -> Result<(), Error> {
374        self.bytes = count.checked_mul(size).and_then(|bytes| self.bytes.checked_add(bytes)).ok_or(Error::Budget)?;
375        if self.bytes > settings.preparation_bytes.get() {
376            return Err(Error::Budget);
377        }
378        Ok(())
379    }
380
381    fn add_work(&mut self, work: usize, settings: &SchemaSettings) -> Result<(), Error> {
382        self.work = self.work.checked_add(work).ok_or(Error::Budget)?;
383        if self.work > settings.preparation_work.get() {
384            return Err(Error::Budget);
385        }
386        Ok(())
387    }
388}
389
390impl<'de> Deserialize<'de> for Schema {
391    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
392        Seed { settings: &SchemaSettings::default() }.deserialize(deserializer)
393    }
394}
395
396impl<'de> DeserializeSeed<'de> for Seed<'_> {
397    type Value = Schema;
398    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Schema, D::Error> {
399        struct Input<'a>(&'a SchemaSettings);
400        impl<'de> Visitor<'de> for Input<'_> {
401            type Value = Schema;
402            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
403                formatter.write_str("a bounded shared message schema")
404            }
405            fn visit_map<A: MapAccess<'de>>(self, mut input: A) -> Result<Schema, A::Error> {
406                let mut root = None;
407                let mut definitions = None;
408                let mut budget = Budget { remaining: self.0.preparation_bytes.get() };
409                while let Some(key) = input.next_key::<String>()? {
410                    match key.as_str() {
411                        "root" => {
412                            if root.is_some() {
413                                return Err(A::Error::duplicate_field("root"));
414                            }
415                            root = Some(input.next_value()?);
416                        }
417                        "definitions" => {
418                            if definitions.is_some() {
419                                return Err(A::Error::duplicate_field("definitions"));
420                            }
421                            definitions = Some(input.next_value_seed(DefinitionsSeed(&mut budget))?);
422                        }
423                        _ => return Err(A::Error::unknown_field(&key, &["root", "definitions"])),
424                    }
425                }
426                let schema = Schema {
427                    root: root.ok_or_else(|| A::Error::missing_field("root"))?,
428                    definitions: definitions.ok_or_else(|| A::Error::missing_field("definitions"))?,
429                };
430                schema.prepare(self.0).map_err(A::Error::custom)?;
431                Ok(schema)
432            }
433        }
434        deserializer.deserialize_map(Input(self.settings))
435    }
436}
437
438impl Budget {
439    fn charge<E: serde::de::Error>(&mut self, bytes: usize) -> Result<(), E> {
440        self.remaining = self.remaining.checked_sub(bytes).ok_or_else(|| E::custom(Error::Budget))?;
441        Ok(())
442    }
443}
444
445impl<'de> DeserializeSeed<'de> for DefinitionsSeed<'_> {
446    type Value = Vec<Vec<Field>>;
447    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
448        deserializer.deserialize_seq(self)
449    }
450}
451
452impl<'de> DeserializeSeed<'de> for DefinitionSeed<'_> {
453    type Value = Vec<Field>;
454    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
455        deserializer.deserialize_seq(self)
456    }
457}
458
459impl<'de> DeserializeSeed<'de> for FieldSeed<'_> {
460    type Value = Field;
461    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> {
462        // Reserve Capacity And Sorting Scratch Before Any Field Allocation.
463        self.0.charge::<D::Error>(4 * size_of::<Field>() + size_of::<&str>() + size_of::<u32>() + MAX_NAME_BYTES)?;
464        Field::deserialize(deserializer)
465    }
466}
467
468impl<'de> Visitor<'de> for DefinitionsSeed<'_> {
469    type Value = Vec<Vec<Field>>;
470    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
471        formatter.write_str("bounded message definitions")
472    }
473    fn visit_seq<A: SeqAccess<'de>>(self, mut input: A) -> Result<Self::Value, A::Error> {
474        let mut definitions = Vec::new();
475        while let Some(fields) = input.next_element_seed(DefinitionSeed(self.0))? {
476            definitions.push(fields);
477        }
478        Ok(definitions)
479    }
480}
481
482impl<'de> Visitor<'de> for DefinitionSeed<'_> {
483    type Value = Vec<Field>;
484    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
485        formatter.write_str("bounded direct message fields")
486    }
487    fn visit_seq<A: SeqAccess<'de>>(self, mut input: A) -> Result<Self::Value, A::Error> {
488        self.0.charge::<A::Error>(4 * size_of::<Vec<Field>>() + size_of::<usize>() + size_of::<bool>())?;
489        let mut fields = Vec::new();
490        while let Some(field) = input.next_element_seed(FieldSeed(self.0))? {
491            fields.push(field);
492        }
493        Ok(fields)
494    }
495}
496
497/* === Public Functions === */
498
499/* === Private Functions === */
500
501fn definition<'de, D: Deserializer<'de>, R: Deserialize<'de>>(deserializer: D) -> Result<Option<R>, D::Error> {
502    R::deserialize(deserializer).map(Some)
503}
504
505fn kind_name<'de, D: Deserializer<'de>>(deserializer: D) -> Result<String, D::Error> {
506    name(deserializer).map(Cow::into_owned)
507}
508
509fn name<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Cow<'static, str>, D::Error> {
510    struct Name;
511    impl<'de> Visitor<'de> for Name {
512        type Value = String;
513        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
514            formatter.write_str("a field name of at most 256 bytes")
515        }
516        fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<String, E> {
517            if value.len() > MAX_NAME_BYTES {
518                return Err(E::custom("schema field name exceeds 256 bytes"));
519            }
520            Ok(value.to_owned())
521        }
522        fn visit_string<E: serde::de::Error>(self, value: String) -> Result<String, E> {
523            if value.len() > MAX_NAME_BYTES {
524                return Err(E::custom("schema field name exceeds 256 bytes"));
525            }
526            Ok(value)
527        }
528    }
529    deserializer.deserialize_string(Name).map(Cow::Owned)
530}
531
532/* === Tests === */
harmos_domain/schema/graph.rs

harmos_domain/schema/
graph.rs
