harmos_storage/
admissions.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::time::UNIX_EPOCH;
18
19// Third Party Imports
20use sha2::{Digest, Sha256};
21
22// Local Imports
23use super::{RecordWrite, StorageCursor, StorageManager, StreamWrite};
24use crate::storage::MutationId;
25use crate::{Error, Identifier, Result};
26
27/* === Constants === */
28
29/* === Macros === */
30
31/* === Type Aliases === */
32
33/* === Structs === */
34
35#[derive(Clone, Copy)]
36pub(super) struct Mutation<'a> {
37    pub(super) id: &'a MutationId,
38    pub(super) fingerprint: [u8; 32],
39}
40
41/* === Enums === */
42
43/* === Traits === */
44
45/* === Implementations === */
46
47impl<C> StorageManager<C> {
48    pub async fn write_schema_once(
49        &self, mutation: &MutationId, key: &super::SchemaKey, schema: &crate::schema::graph::Schema,
50    ) -> Result<StorageCursor> {
51        self.admit_schema(key, schema, Some(mutation)).await
52    }
53
54    pub async fn write_records_once(&self, mutation: &MutationId, records: &[RecordWrite<'_>]) -> Result<StorageCursor> {
55        self.admit_record_values(mutation, records.iter().copied()).await
56    }
57
58    pub(crate) async fn admit_record_values<'a>(
59        &self, mutation: &MutationId, records: impl ExactSizeIterator<Item = RecordWrite<'a>> + Clone,
60    ) -> Result<StorageCursor> {
61        nonempty(records.len())?;
62        let mut hash = Sha256::new();
63        part(&mut hash, b"storage.write_records:v1");
64        hash.update((records.len() as u64).to_le_bytes());
65        for record in records.clone() {
66            hash_record(&mut hash, &record);
67        }
68        self.admit_records(records, Some(Mutation { id: mutation, fingerprint: hash.finalize().into() })).await
69    }
70
71    pub async fn write_stream_once(&self, mutation: &MutationId, rows: &[StreamWrite<'_>]) -> Result<StorageCursor> {
72        self.admit_stream_values(mutation, rows.iter().copied()).await
73    }
74
75    pub(crate) async fn admit_stream_values<'a>(
76        &self, mutation: &MutationId, rows: impl ExactSizeIterator<Item = StreamWrite<'a>> + Clone,
77    ) -> Result<StorageCursor> {
78        nonempty(rows.len())?;
79        let mut hash = Sha256::new();
80        part(&mut hash, b"storage.write_stream:v1");
81        hash.update((rows.len() as u64).to_le_bytes());
82        for row in rows.clone() {
83            part(&mut hash, row.stream.as_str().as_bytes());
84            hash.update(row.sequence.to_le_bytes());
85            hash_record(&mut hash, &row.record);
86        }
87        self.admit_stream(rows, Some(Mutation { id: mutation, fingerprint: hash.finalize().into() })).await
88    }
89}
90
91/* === Public Functions === */
92
93/* === Private Functions === */
94
95// Called After Shared Schema Preflight Has Bounded The Finite Definition Table.
96pub(super) fn schema_mutation<'a>(
97    id: &'a MutationId, key: &super::SchemaKey, schema: &crate::schema::graph::Schema,
98) -> Result<Mutation<'a>> {
99    let mut hash = Sha256::new();
100    part(&mut hash, b"storage.write_schema:v1");
101    part(&mut hash, key.id.as_str().as_bytes());
102    hash.update(key.version.get().to_le_bytes());
103    hash.update((schema.root as u64).to_le_bytes());
104    hash.update((schema.definitions.len() as u64).to_le_bytes());
105    for fields in &schema.definitions {
106        hash_fields(&mut hash, fields)?;
107    }
108    Ok(Mutation { id, fingerprint: hash.finalize().into() })
109}
110fn hash_fields(hash: &mut Sha256, fields: &[crate::schema::graph::Field]) -> Result<()> {
111    use crate::schema::Cardinality;
112    use crate::schema::graph::Kind;
113    hash.update((fields.len() as u64).to_le_bytes());
114    for field in fields {
115        if field.name.len() > crate::schema::MAX_NAME_BYTES {
116            return Err(Error::InvalidInput { field: "storage schema", reason: "field name exceeds limit" });
117        }
118        part(hash, field.name.as_bytes());
119        hash.update(field.tag.to_le_bytes());
120        hash.update([match field.cardinality {
121            Cardinality::Required => 0,
122            Cardinality::Optional => 1,
123            Cardinality::Repeated => 2,
124            Cardinality::Variant => 3,
125        }]);
126        hash.update([match field.kind {
127            Kind::Bool => 0,
128            Kind::U32 => 1,
129            Kind::U64 => 2,
130            Kind::I32 => 3,
131            Kind::I64 => 4,
132            Kind::F32 => 5,
133            Kind::F64 => 6,
134            Kind::String => 7,
135            Kind::Bytes => 8,
136            Kind::Message { .. } => 9,
137        }]);
138        if let Kind::Message { definition } = field.kind {
139            hash.update((definition as u64).to_le_bytes());
140        }
141    }
142    Ok(())
143}
144
145fn nonempty(count: usize) -> Result<()> {
146    if count == 0 {
147        return Err(Error::InvalidInput { field: "storage mutation", reason: "write batch must not be empty" });
148    }
149    Ok(())
150}
151fn part(hash: &mut Sha256, bytes: &[u8]) {
152    hash.update((bytes.len() as u64).to_le_bytes());
153    hash.update(bytes);
154}
155fn optional(hash: &mut Sha256, value: Option<&Identifier>) {
156    hash.update([u8::from(value.is_some())]);
157    if let Some(value) = value {
158        part(hash, value.as_str().as_bytes());
159    }
160}
161fn hash_record(hash: &mut Sha256, record: &RecordWrite<'_>) {
162    part(hash, record.schema.id.as_str().as_bytes());
163    hash.update(record.schema.version.get().to_le_bytes());
164    optional(hash, record.scope);
165    part(hash, record.item.as_str().as_bytes());
166    optional(hash, record.source);
167    hash.update([u8::from(record.source_time.is_some())]);
168    if let Some(time) = record.source_time {
169        let (negative, duration) = match time.duration_since(UNIX_EPOCH) {
170            Ok(duration) => (false, duration),
171            Err(error) => (true, error.duration()),
172        };
173        hash.update([u8::from(negative)]);
174        hash.update(duration.as_secs().to_le_bytes());
175        hash.update(duration.subsec_nanos().to_le_bytes());
176    }
177    part(hash, record.payload);
178}
179
180/* === Tests === */
harmos_storage/admissions.rs

harmos_storage/
admissions.rs
