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

harmos_storage/
admissions.rs
