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

harmos_runtime/records/
registration.rs
