harmos_runtime/records/
registration.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::collections::BTreeMap;
18use std::sync::{Arc, Mutex};
19// Third Party Imports
20
21// Local Imports
22use crate::schema::{Validator, graph::Schema};
23use crate::storage::{ErasedCatalog, SchemaKey, StorageManager};
24use crate::{Error, MemorySize, Result};
25
26/* === Constants === */
27
28const DEFINITIONS: usize = 64;
29const REGISTRY_BYTES: usize = 8 * 1024 * 1024;
30const NAME_BYTES: usize = 1024;
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36type Key = (String, u32);
37
38/* === Structs === */
39
40#[derive(Debug)]
41pub struct Definition {
42    owner: String,
43    schema: SchemaKey,
44    graph: Schema,
45    validation: Validator,
46}
47
48pub(crate) struct Registry {
49    pub(super) storage: Option<StorageManager<ErasedCatalog>>,
50    state: Mutex<State>,
51
52    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
53    pub(super) listeners: Mutex<BTreeMap<String, Arc<super::listeners::Listening>>>,
54}
55
56#[derive(Default)]
57struct State {
58    definitions: BTreeMap<Key, Entry>,
59    bytes: usize,
60    constructing_bytes: usize,
61    constructing: usize,
62}
63
64struct Entry {
65    definition: Arc<Definition>,
66    bytes: usize,
67    ready: bool,
68}
69
70struct Construction {
71    registry: Arc<Registry>,
72    bytes: usize,
73}
74
75pub(super) struct Pending {
76    registry: Arc<Registry>,
77    key: Key,
78    pub(super) definition: Arc<Definition>,
79    committed: bool,
80}
81
82/* === Enums === */
83
84/* === Traits === */
85
86/* === Implementations === */
87
88impl Definition {
89    pub fn owner(&self) -> &str {
90        &self.owner
91    }
92
93    pub fn schema(&self) -> &SchemaKey {
94        &self.schema
95    }
96
97    pub fn graph(&self) -> &Schema {
98        &self.graph
99    }
100
101    pub fn validate(&self, payload: &[u8]) -> Result<()> {
102        self.validation.validate(payload).map_err(|source| Error::RecordPayload { source })
103    }
104}
105
106impl Registry {
107    pub(crate) fn new(storage: Option<StorageManager<ErasedCatalog>>) -> Self {
108        Self {
109            storage,
110            state: Mutex::new(State::default()),
111            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
112            listeners: Mutex::new(BTreeMap::new()),
113        }
114    }
115
116    pub(super) fn reserve(self: &Arc<Self>, owner: &str, schema: &SchemaKey, graph: &Schema) -> Result<Pending> {
117        schema.require_postcard()?;
118        if !harmos_domain::artifact_subset(owner) || owner.len() > NAME_BYTES || schema.id.as_str().len() > NAME_BYTES {
119            return Err(Error::InvalidInput { field: "record schema", reason: "invalid or oversized identity" });
120        }
121        let settings = harmos_domain::config::SchemaSettings::default();
122        graph.preparation(&settings).map_err(|_| limit())?;
123        let fields_bytes = graph.clone_memory_bytes().map_err(|_| limit())?.checked_sub(std::mem::size_of::<Schema>()).ok_or_else(limit)?;
124        {
125            let state = self.state.lock().expect("record schema registry poisoned");
126            if let Some((key, entry)) =
127                state.definitions.iter().find(|((id, version), _)| id == schema.id.as_str() && *version == schema.version.get())
128            {
129                if entry.ready && entry.definition.owner == owner && entry.definition.graph == *graph {
130                    return Ok(Pending {
131                        registry: Arc::clone(self),
132                        key: key.clone(),
133                        definition: Arc::clone(&entry.definition),
134                        committed: true,
135                    });
136                }
137                return Err(Error::AlreadyRegistered { domain: "record schema", name: key.0.clone() });
138            }
139            if state.definitions.values().any(|entry| entry.definition.schema.id == schema.id && entry.definition.owner != owner) {
140                return Err(Error::AlreadyRegistered { domain: "record schema owner", name: schema.id.as_str().to_owned() });
141            }
142        }
143        let destination = [
144            std::mem::size_of::<Definition>(),
145            std::mem::size_of::<(Key, Entry)>(),
146            1024,
147            owner.len(),
148            schema.id.as_str().len(),
149            schema.id.as_str().len(),
150            fields_bytes,
151        ]
152        .into_iter()
153        .try_fold(0usize, usize::checked_add)
154        .ok_or_else(limit)?;
155        let compilation = Validator::construction_memory_bytes(graph, &settings)
156            .map_err(|source| Error::RecordSchema { source: crate::schema::Error::Graph(source) })?;
157        let construction_bytes = destination.checked_add(compilation).ok_or_else(limit)?;
158        let mut construction = Construction::new(self, construction_bytes)?;
159        // Compile Outside The Registry Lock Under The Original Shared Budget
160        let validation = Validator::from_schema(graph, &settings)
161            .map_err(|source| Error::RecordSchema { source: crate::schema::Error::Graph(source) })?;
162        let bytes = destination.checked_add(validation.memory_size()).ok_or_else(limit)?;
163        if bytes > construction_bytes {
164            return Err(limit());
165        }
166        let key = (schema.id.as_str().to_owned(), schema.version.get());
167        let definition = Arc::new(Definition { owner: owner.to_owned(), schema: schema.clone(), graph: graph.clone(), validation });
168        let mut state = self.state.lock().expect("record schema registry poisoned");
169        if state.definitions.contains_key(&key) {
170            return Err(Error::AlreadyRegistered { domain: "record schema", name: key.0 });
171        }
172        if state.definitions.values().any(|entry| entry.definition.schema.id == schema.id && entry.definition.owner != owner) {
173            return Err(Error::AlreadyRegistered { domain: "record schema owner", name: key.0 });
174        }
175        construction.transfer(&mut state, bytes);
176        state.definitions.insert(key.clone(), Entry { definition: Arc::clone(&definition), bytes, ready: false });
177        Ok(Pending { registry: Arc::clone(self), key, definition, committed: false })
178    }
179
180    pub(super) fn definition(&self, schema: &SchemaKey) -> Option<Arc<Definition>> {
181        if schema.id.as_str().len() > NAME_BYTES || schema.require_postcard().is_err() {
182            return None;
183        }
184        let state = self.state.lock().expect("record schema registry poisoned");
185        state
186            .definitions
187            .get(&(schema.id.as_str().to_owned(), schema.version.get()))
188            .filter(|entry| entry.ready)
189            .map(|entry| Arc::clone(&entry.definition))
190    }
191
192    pub(super) fn definitions(&self) -> Vec<Arc<Definition>> {
193        self.state
194            .lock()
195            .expect("record schema registry poisoned")
196            .definitions
197            .values()
198            .filter(|entry| entry.ready)
199            .map(|entry| Arc::clone(&entry.definition))
200            .collect()
201    }
202}
203
204impl Construction {
205    fn new(registry: &Arc<Registry>, bytes: usize) -> Result<Self> {
206        let mut state = registry.state.lock().expect("record schema registry poisoned");
207        let available = REGISTRY_BYTES.saturating_sub(state.bytes).saturating_sub(state.constructing_bytes);
208        if state.definitions.len() + state.constructing >= DEFINITIONS || bytes > available {
209            return Err(limit());
210        }
211        state.constructing_bytes += bytes;
212        state.constructing += 1;
213        Ok(Self { registry: Arc::clone(registry), bytes })
214    }
215
216    fn transfer(&mut self, state: &mut State, bytes: usize) {
217        state.constructing_bytes -= self.bytes;
218        state.constructing -= 1;
219        state.bytes += bytes;
220        self.bytes = 0;
221    }
222}
223
224impl Drop for Construction {
225    fn drop(&mut self) {
226        if self.bytes == 0 {
227            return;
228        }
229        let mut state = self.registry.state.lock().expect("record schema registry poisoned");
230        state.constructing_bytes -= self.bytes;
231        state.constructing -= 1;
232    }
233}
234
235impl Pending {
236    pub(super) fn is_committed(&self) -> bool {
237        self.committed
238    }
239
240    pub(super) fn commit(mut self) -> Arc<Definition> {
241        if !self.committed {
242            let mut state = self.registry.state.lock().expect("record schema registry poisoned");
243            let entry = state.definitions.get_mut(&self.key).expect("reserved schema remains owned");
244            entry.ready = true;
245            self.committed = true;
246        }
247        Arc::clone(&self.definition)
248    }
249}
250
251impl Drop for Pending {
252    fn drop(&mut self) {
253        if self.committed {
254            return;
255        }
256        let removed = {
257            let mut state = self.registry.state.lock().expect("record schema registry poisoned");
258            let Some(entry) = state.definitions.get(&self.key) else {
259                return;
260            };
261            if !Arc::ptr_eq(&entry.definition, &self.definition) {
262                return;
263            }
264            let entry = state.definitions.remove(&self.key).expect("reserved schema exists");
265            state.bytes -= entry.bytes;
266            entry
267        };
268        drop(removed);
269    }
270}
271
272/* === Public Functions === */
273
274/* === Private Functions === */
275
276fn limit() -> Error {
277    Error::ResourceExhausted { resource: "record schema registry" }
278}
279
280/* === Tests === */
harmos_runtime/records/registration.rs

harmos_runtime/records/
registration.rs
