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: Arc<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(crate) fn graph_owner(&self) -> Arc<Schema> {
102        Arc::clone(&self.graph)
103    }
104
105    pub fn validate(&self, payload: &[u8]) -> Result<()> {
106        self.validation.validate(payload).map_err(|source| Error::RecordPayload { source })
107    }
108}
109
110impl Registry {
111    pub(crate) fn new(storage: Option<StorageManager<ErasedCatalog>>) -> Self {
112        Self {
113            storage,
114            state: Mutex::new(State::default()),
115            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
116            listeners: Mutex::new(BTreeMap::new()),
117        }
118    }
119
120    pub(super) fn reserve(self: &Arc<Self>, owner: &str, schema: SchemaKey, graph: Arc<Schema>) -> Result<Pending> {
121        schema.require_postcard()?;
122        if !harmos_domain::artifact_subset(owner) || owner.len() > NAME_BYTES || schema.id.as_str().len() > NAME_BYTES {
123            return Err(Error::InvalidInput { field: "record schema", reason: "invalid or oversized identity" });
124        }
125        let settings = harmos_domain::config::SchemaSettings::default();
126        graph.preparation(&settings).map_err(|_| limit())?;
127        let fields_bytes = graph.memory_size().checked_add(2 * std::mem::size_of::<usize>()).ok_or_else(limit)?;
128        {
129            let state = self.state.lock().expect("record schema registry poisoned");
130            if let Some((key, entry)) =
131                state.definitions.iter().find(|((id, version), _)| id == schema.id.as_str() && *version == schema.version.get())
132            {
133                if entry.ready && entry.definition.owner == owner && entry.definition.graph == graph {
134                    return Ok(Pending {
135                        registry: Arc::clone(self),
136                        key: key.clone(),
137                        definition: Arc::clone(&entry.definition),
138                        committed: true,
139                    });
140                }
141                return Err(Error::AlreadyRegistered { domain: "record schema", name: key.0.clone() });
142            }
143            if state.definitions.values().any(|entry| entry.definition.schema.id == schema.id && entry.definition.owner != owner) {
144                return Err(Error::AlreadyRegistered { domain: "record schema owner", name: schema.id.as_str().to_owned() });
145            }
146        }
147        let destination = [
148            std::mem::size_of::<Definition>(),
149            std::mem::size_of::<(Key, Entry)>(),
150            1024,
151            owner.len(),
152            schema.id.as_str().len(),
153            schema.id.as_str().len(),
154            fields_bytes,
155        ]
156        .into_iter()
157        .try_fold(0usize, usize::checked_add)
158        .ok_or_else(limit)?;
159        let compilation = Validator::construction_memory_bytes(&graph, &settings)
160            .map_err(|source| Error::RecordSchema { source: crate::schema::Error::Graph(source) })?;
161        let construction_bytes = destination.checked_add(compilation).ok_or_else(limit)?;
162        let mut construction = Construction::new(self, construction_bytes)?;
163        // Compile Outside The Registry Lock Under The Original Shared Budget
164        let validation = Validator::from_schema(&graph, &settings)
165            .map_err(|source| Error::RecordSchema { source: crate::schema::Error::Graph(source) })?;
166        let bytes = destination.checked_add(validation.memory_size()).ok_or_else(limit)?;
167        if bytes > construction_bytes {
168            return Err(limit());
169        }
170        let key = (schema.id.as_str().to_owned(), schema.version.get());
171        let definition = Arc::new(Definition { owner: owner.to_owned(), schema: schema.clone(), graph, validation });
172        let mut state = self.state.lock().expect("record schema registry poisoned");
173        if state.definitions.contains_key(&key) {
174            return Err(Error::AlreadyRegistered { domain: "record schema", name: key.0 });
175        }
176        if state.definitions.values().any(|entry| entry.definition.schema.id == schema.id && entry.definition.owner != owner) {
177            return Err(Error::AlreadyRegistered { domain: "record schema owner", name: key.0 });
178        }
179        construction.transfer(&mut state, bytes);
180        state.definitions.insert(key.clone(), Entry { definition: Arc::clone(&definition), bytes, ready: false });
181        Ok(Pending { registry: Arc::clone(self), key, definition, committed: false })
182    }
183
184    pub(super) fn definition(&self, schema: &SchemaKey) -> Option<Arc<Definition>> {
185        if schema.id.as_str().len() > NAME_BYTES || schema.require_postcard().is_err() {
186            return None;
187        }
188        let state = self.state.lock().expect("record schema registry poisoned");
189        state
190            .definitions
191            .get(&(schema.id.as_str().to_owned(), schema.version.get()))
192            .filter(|entry| entry.ready)
193            .map(|entry| Arc::clone(&entry.definition))
194    }
195
196    pub(super) fn definitions(&self) -> Vec<Arc<Definition>> {
197        self.state
198            .lock()
199            .expect("record schema registry poisoned")
200            .definitions
201            .values()
202            .filter(|entry| entry.ready)
203            .map(|entry| Arc::clone(&entry.definition))
204            .collect()
205    }
206}
207
208impl Construction {
209    fn new(registry: &Arc<Registry>, bytes: usize) -> Result<Self> {
210        let mut state = registry.state.lock().expect("record schema registry poisoned");
211        let available = REGISTRY_BYTES.saturating_sub(state.bytes).saturating_sub(state.constructing_bytes);
212        if state.definitions.len() + state.constructing >= DEFINITIONS || bytes > available {
213            return Err(limit());
214        }
215        state.constructing_bytes += bytes;
216        state.constructing += 1;
217        Ok(Self { registry: Arc::clone(registry), bytes })
218    }
219
220    fn transfer(&mut self, state: &mut State, bytes: usize) {
221        state.constructing_bytes -= self.bytes;
222        state.constructing -= 1;
223        state.bytes += bytes;
224        self.bytes = 0;
225    }
226}
227
228impl Drop for Construction {
229    fn drop(&mut self) {
230        if self.bytes == 0 {
231            return;
232        }
233        let mut state = self.registry.state.lock().expect("record schema registry poisoned");
234        state.constructing_bytes -= self.bytes;
235        state.constructing -= 1;
236    }
237}
238
239impl Pending {
240    pub(super) fn is_committed(&self) -> bool {
241        self.committed
242    }
243
244    pub(super) fn commit(mut self) -> Arc<Definition> {
245        if !self.committed {
246            let mut state = self.registry.state.lock().expect("record schema registry poisoned");
247            let entry = state.definitions.get_mut(&self.key).expect("reserved schema remains owned");
248            entry.ready = true;
249            self.committed = true;
250        }
251        Arc::clone(&self.definition)
252    }
253}
254
255impl Drop for Pending {
256    fn drop(&mut self) {
257        if self.committed {
258            return;
259        }
260        let removed = {
261            let mut state = self.registry.state.lock().expect("record schema registry poisoned");
262            let Some(entry) = state.definitions.get(&self.key) else {
263                return;
264            };
265            if !Arc::ptr_eq(&entry.definition, &self.definition) {
266                return;
267            }
268            let entry = state.definitions.remove(&self.key).expect("reserved schema exists");
269            state.bytes -= entry.bytes;
270            entry
271        };
272        drop(removed);
273    }
274}
275
276/* === Public Functions === */
277
278/* === Private Functions === */
279
280fn limit() -> Error {
281    Error::ResourceExhausted { resource: "record schema registry" }
282}
283
284/* === Tests === */
harmos_runtime/records/registration.rs

harmos_runtime/records/
registration.rs
