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::{RecordInput, RecordWrite, StorageCursor, StorageManager, StreamInput};
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    #[tracing::instrument(name = "storage.admissions.write_schema_once", level = "debug", skip_all)]
49    pub async fn write_schema_once(
50        &self, mutation: &MutationId, key: super::SchemaKey, schema: impl Into<std::sync::Arc<crate::schema::graph::Schema>>,
51    ) -> Result<StorageCursor> {
52        self.admit_schema(key, schema.into(), Some(mutation)).await
53    }
54
55    #[tracing::instrument(name = "storage.admissions.write_records_once", level = "debug", skip_all)]
56    pub async fn write_records_once(&self, mutation: &MutationId, records: Vec<RecordInput>) -> Result<StorageCursor> {
57        self.admit_record_values(mutation, records).await
58    }
59
60    #[tracing::instrument(name = "storage.admissions.admit_record_values", level = "debug", skip_all)]
61    pub(crate) async fn admit_record_values(&self, mutation: &MutationId, records: Vec<RecordInput>) -> Result<StorageCursor> {
62        nonempty(records.len())?;
63        let mut hash = Sha256::new();
64        part(&mut hash, b"storage.write_records:v1");
65        hash.update((records.len() as u64).to_le_bytes());
66        for record in &records {
67            hash_record(&mut hash, &record.borrowed());
68        }
69        self.admit_records(
70            records,
71            Some(Mutation {
72                id: mutation,
73                fingerprint: hash.finalize().into(),
74            }),
75        )
76        .await
77    }
78
79    #[tracing::instrument(name = "storage.admissions.write_stream_once", level = "debug", skip_all)]
80    pub async fn write_stream_once(&self, mutation: &MutationId, rows: Vec<StreamInput>) -> Result<StorageCursor> {
81        self.admit_stream_values(mutation, rows).await
82    }
83
84    #[tracing::instrument(name = "storage.admissions.admit_stream_values", level = "debug", skip_all)]
85    pub(crate) async fn admit_stream_values(&self, mutation: &MutationId, rows: Vec<StreamInput>) -> Result<StorageCursor> {
86        nonempty(rows.len())?;
87        let mut hash = Sha256::new();
88        part(&mut hash, b"storage.write_stream:v1");
89        hash.update((rows.len() as u64).to_le_bytes());
90        for row in &rows {
91            part(&mut hash, row.stream.as_str().as_bytes());
92            hash.update(row.sequence.to_le_bytes());
93            hash_record(&mut hash, &row.record.borrowed());
94        }
95        self.admit_stream(
96            rows,
97            Some(Mutation {
98                id: mutation,
99                fingerprint: hash.finalize().into(),
100            }),
101        )
102        .await
103    }
104}
105
106/* === Public Functions === */
107
108/* === Private Functions === */
109
110// Called After Shared Schema Preflight Has Bounded The Finite Definition Table.
111pub(super) fn schema_mutation<'a>(
112    id: &'a MutationId, key: &super::SchemaKey, schema: &crate::schema::graph::Schema,
113) -> Result<Mutation<'a>> {
114    let mut hash = Sha256::new();
115    part(&mut hash, b"storage.write_schema:v1");
116    part(&mut hash, key.id.as_str().as_bytes());
117    hash.update(key.version.get().to_le_bytes());
118    hash.update((schema.root as u64).to_le_bytes());
119    hash.update((schema.definitions.len() as u64).to_le_bytes());
120    for fields in &schema.definitions {
121        hash_fields(&mut hash, fields)?;
122    }
123    Ok(Mutation { id, fingerprint: hash.finalize().into() })
124}
125fn hash_fields(hash: &mut Sha256, fields: &[crate::schema::graph::Field]) -> Result<()> {
126    use crate::schema::Cardinality;
127    use crate::schema::graph::Kind;
128    hash.update((fields.len() as u64).to_le_bytes());
129    for field in fields {
130        if field.name.len() > crate::schema::MAX_NAME_BYTES {
131            return Err(Error::InvalidInput {
132                field: "storage schema",
133                reason: "field name exceeds limit",
134            });
135        }
136        part(hash, field.name.as_bytes());
137        hash.update(field.tag.to_le_bytes());
138        hash.update([match field.cardinality {
139            Cardinality::Required => 0,
140            Cardinality::Optional => 1,
141            Cardinality::Repeated => 2,
142            Cardinality::Variant => 3,
143        }]);
144        hash.update([match field.kind {
145            Kind::Bool => 0,
146            Kind::U32 => 1,
147            Kind::U64 => 2,
148            Kind::I32 => 3,
149            Kind::I64 => 4,
150            Kind::F32 => 5,
151            Kind::F64 => 6,
152            Kind::String => 7,
153            Kind::Bytes => 8,
154            Kind::Message { .. } => 9,
155        }]);
156        if let Kind::Message { definition } = field.kind {
157            hash.update((definition as u64).to_le_bytes());
158        }
159    }
160    Ok(())
161}
162
163fn nonempty(count: usize) -> Result<()> {
164    if count == 0 {
165        return Err(Error::InvalidInput {
166            field: "storage mutation",
167            reason: "write batch must not be empty",
168        });
169    }
170    Ok(())
171}
172fn part(hash: &mut Sha256, bytes: &[u8]) {
173    hash.update((bytes.len() as u64).to_le_bytes());
174    hash.update(bytes);
175}
176fn optional(hash: &mut Sha256, value: Option<&Identifier>) {
177    hash.update([u8::from(value.is_some())]);
178    if let Some(value) = value {
179        part(hash, value.as_str().as_bytes());
180    }
181}
182fn hash_record(hash: &mut Sha256, record: &RecordWrite<'_>) {
183    part(hash, record.schema.id.as_str().as_bytes());
184    hash.update(record.schema.version.get().to_le_bytes());
185    optional(hash, record.scope);
186    part(hash, record.item.as_str().as_bytes());
187    optional(hash, record.source);
188    hash.update([u8::from(record.source_time.is_some())]);
189    if let Some(time) = record.source_time {
190        let (negative, duration) = match time.duration_since(UNIX_EPOCH) {
191            Ok(duration) => (false, duration),
192            Err(error) => (true, error.duration()),
193        };
194        hash.update([u8::from(negative)]);
195        hash.update(duration.as_secs().to_le_bytes());
196        hash.update(duration.subsec_nanos().to_le_bytes());
197    }
198    part(hash, record.payload);
199}
200
201/* === Tests === */
harmos_storage/admissions.rs

harmos_storage/
admissions.rs
