harmos_runtime/operations/
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::sync::Arc;
18// Third Party Imports
19
20// Local Imports
21#[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
22use super::Binding;
23use super::Manager;
24#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
25use super::manager::Entry;
26#[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
27use crate::catalog::Projection;
28#[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
29use crate::catalog::{Catalog, Visitor};
30
31#[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
32use crate::operations::Declarations;
33use crate::operations::OperationDeclaration;
34#[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
35use crate::operations::OperationDefinition;
36#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
37use crate::plugin::PluginScope;
38
39#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
40use crate::Error;
41use crate::{Result, operations::Lane};
42
43/* === Constants === */
44
45/* === Macros === */
46
47/* === Type Aliases === */
48
49type BindCatalog = dyn Fn(&Manager) -> Result<()> + Send + Sync;
50
51/* === Structs === */
52
53#[derive(Clone)]
54pub struct Registration<L: Lane> {
55    pub(crate) batch: harmos_sdk::RegistrationBatch,
56    pub(crate) lane: Option<L>,
57    pub(crate) defaults: std::collections::BTreeMap<&'static str, L>,
58}
59
60#[derive(Clone)]
61pub(crate) struct CatalogBinding {
62    bind: Arc<BindCatalog>,
63}
64
65#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
66pub struct Bindings<'a> {
67    pub(super) operations: &'a Manager,
68    owner: &'a str,
69
70    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
71    pub(super) scope: Option<&'a PluginScope>,
72
73    pub(super) entries: Vec<Arc<Entry>>,
74}
75
76#[must_use = "retain plugin operation registration until shutdown settles its invocations"]
77#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
78pub struct PluginRegistration {
79    scope: Arc<PluginScope>,
80    operations: Manager,
81    entries: Vec<Arc<Entry>>,
82}
83
84/* === Enums === */
85
86/* === Traits === */
87
88pub trait RegistrationExt {
89    fn lane<L: Lane>(self, lane: L) -> Registration<L>;
90    fn defaults<L: Lane>(self) -> Registration<L>;
91}
92
93/* === Implementations === */
94
95impl RegistrationExt for harmos_sdk::RegistrationBatch {
96    fn lane<L: Lane>(self, lane: L) -> Registration<L> {
97        Registration {
98            batch: self,
99            lane: Some(lane),
100            defaults: std::collections::BTreeMap::new(),
101        }
102    }
103
104    fn defaults<L: Lane>(self) -> Registration<L> {
105        Registration {
106            batch: self,
107            lane: None,
108            defaults: std::collections::BTreeMap::new(),
109        }
110    }
111}
112
113impl<L: Lane> Registration<L> {
114    pub fn default_lane_for<O: OperationDeclaration>(mut self, lane: L) -> Self {
115        self.defaults.insert(O::ID, lane);
116        self
117    }
118}
119
120impl CatalogBinding {
121    pub(crate) fn new<L: Lane>(registration: Registration<L>) -> Self {
122        Self {
123            bind: Arc::new(move |operations| operations.register_native(registration.clone())),
124        }
125    }
126    pub(crate) fn bind(&self, operations: &Manager) -> Result<()> {
127        (self.bind)(operations)
128    }
129}
130
131impl Manager {
132    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
133    pub fn stage_plugin(&self, scope: &PluginScope, configure: impl FnOnce(&mut Bindings<'_>) -> Result<()>) -> Result<PluginRegistration> {
134        if scope.is_closed() {
135            return Err(Error::Closing { domain: "plugin scope" });
136        }
137        if scope.has_committed() {
138            return Err(Error::InvalidInput {
139                field: "operation catalog",
140                reason: "plugin generation is already committed",
141            });
142        }
143        let mut bindings = Bindings {
144            operations: self,
145            owner: scope.owner(),
146            scope: Some(scope),
147            entries: Vec::new(),
148        };
149        configure(&mut bindings)?;
150        scope.validate_operations(bindings.entries.iter().map(|entry| entry.declared()))?;
151        scope.validate_wire_contracts(bindings.entries.iter().filter_map(|entry| entry.contract.as_ref()))?;
152        self.reserve(&bindings.entries)?;
153        #[cfg(feature = "plugin-sidecar")]
154        for entry in &bindings.entries {
155            if let Some(native) = &entry.native
156                && let Err(error) = native.capture_observe_contract()
157            {
158                self.fence(&bindings.entries, true);
159                self.remove_settled(&bindings.entries);
160                return Err(error);
161            }
162        }
163        let receipt_scope = Arc::new(scope.clone());
164        let staged = self
165            .track_staging(&receipt_scope)
166            .and_then(|_| self.validate_staging(&bindings.entries))
167            .and_then(|_| scope.bind_operations())
168            .and_then(|_| self.validate_staging(&bindings.entries));
169        if let Err(error) = staged {
170            scope.close_staging();
171            self.fence(&bindings.entries, true);
172            self.remove_settled(&bindings.entries);
173            return Err(error);
174        }
175        Ok(PluginRegistration {
176            scope: receipt_scope,
177            operations: self.clone(),
178            entries: bindings.entries,
179        })
180    }
181}
182
183#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
184impl Bindings<'_> {
185    pub fn register<L: Lane>(&mut self, registration: Registration<L>) -> Result<()> {
186        let entries = self.operations.native_entries(registration, self.owner, self.scope)?;
187        self.entries.extend(entries);
188        Ok(())
189    }
190
191    #[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
192    pub(super) fn stage_implementation<D, O>(&mut self, lane: &'static str, dependencies: harmos_sdk::Dependencies) -> Result<()>
193    where
194        D: OperationDeclaration,
195        O: harmos_sdk::operations::Carrier<Declaration = D>,
196    {
197        let contract = harmos_sdk::operations::WireDeclaration::of::<D>()
198            .contract_with_settings(&self.operations.schema_settings)
199            .map_err(|source| Error::OperationFailed {
200                operation: D::ID.to_owned(),
201                source: Arc::new(source),
202            })?;
203        let wire = (contract.input().is_none() && contract.output().is_none())
204            .then_some(crate::operations::wire::dispatch_native as crate::operations::wire::Dispatch);
205        let mut batch = harmos_sdk::RegistrationBatch::new(dependencies).carrier::<D, O>();
206        let factory = batch.operations.pop().expect("one native implementation");
207        let mut entry = self.operations.binding_erased(self.owner, lane, Arc::new(factory), D::definition(), wire, self.scope.cloned())?;
208        #[cfg(feature = "grpc")]
209        if contract.grpc().is_some() {
210            Arc::get_mut(&mut entry).expect("new native entry").grpc = Some(crate::grpc::routing::handler::<D>());
211        }
212        Arc::get_mut(&mut entry).expect("new native entry").contract = Some(contract);
213        self.entries.push(entry);
214        Ok(())
215    }
216}
217
218#[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
219impl Bindings<'_> {
220    #[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
221    pub(super) fn project<C, V, H>(&mut self, lane: impl Lane, handler: Arc<H>) -> Result<()>
222    where
223        C: Catalog<Declarations> + Catalog<V>,
224        V: Visitor<Entry = Binding<H>>,
225        H: Send + Sync + 'static,
226    {
227        self.project_at::<C, V, H>(lane.name(), handler)
228    }
229
230    #[cfg(any(feature = "plugin-sidecar", all(feature = "plugin-guest", not(target_arch = "wasm32"))))]
231    fn project_at<C, V, H>(&mut self, lane: &'static str, handler: Arc<H>) -> Result<()>
232    where
233        C: Catalog<Declarations> + Catalog<V>,
234        V: Visitor<Entry = Binding<H>>,
235        H: Send + Sync + 'static,
236    {
237        let definitions = Declarations::of::<C>().map_err(|source| Error::OperationDeclaration { source })?;
238        self.apply_projection(lane, handler, &definitions, &<C as Catalog<V>>::ENTRIES)
239    }
240
241    fn apply_projection<H: Send + Sync + 'static>(
242        &mut self, lane: &'static str, handler: Arc<H>, definitions: &[OperationDefinition], entries: &impl Projection<Binding<H>>,
243    ) -> Result<()> {
244        let previous = self.entries.len();
245        if let Err(error) = entries.try_for_each(|entry| entry.apply(self, lane, Arc::clone(&handler))) {
246            self.entries.truncate(previous);
247            return Err(error);
248        }
249        let projected = &self.entries[previous..];
250        if projected.len() != definitions.len() || projected.iter().zip(definitions).any(|(entry, declared)| entry.declared() != declared) {
251            self.entries.truncate(previous);
252            return Err(Error::InvalidInput {
253                field: "operation catalog",
254                reason: "metadata and handler projections differ",
255            });
256        }
257        Ok(())
258    }
259}
260
261#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
262impl PluginRegistration {
263    pub async fn shutdown(&mut self) -> Result<()> {
264        self.scope.close();
265        self.entries = self.operations.scope_entries(&self.scope);
266        self.scope.unbind_operations();
267        self.operations.fence(&self.entries, false);
268        self.operations.settle_entries(&self.entries).await?;
269        self.operations.resident.shutdown_plugin(&self.scope).await?;
270        self.entries.clear();
271        Ok(())
272    }
273}
274
275#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
276impl Drop for PluginRegistration {
277    fn drop(&mut self) {
278        self.scope.close();
279        self.entries = self.operations.scope_entries(&self.scope);
280        self.scope.unbind_operations();
281        self.operations.fence(&self.entries, true);
282        self.operations.remove_settled(&self.entries);
283        // Running entries remain fenced and reserved in the manager. A later
284        // reservation prunes settled tombstones; runtime shutdown drains all.
285    }
286}
287
288/* === Public Functions === */
289
290/* === Private Functions === */
291
292/* === Tests === */
harmos_runtime/operations/registration.rs

harmos_runtime/operations/
registration.rs
