harmos_runtime/operations/manager/
construction.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
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl OperationManager {
38    pub(crate) fn new(
39        instances: Arc<Instances>, settings: &crate::operations::Settings, reserved: std::collections::BTreeSet<String>, identity: Arc<()>,
40    ) -> Self {
41        Self {
42            resident: crate::operations::resident::Registry::new(&instances, reserved, identity),
43            context: Arc::new(std::sync::OnceLock::from(crate::operations::CallContext::default())),
44            instances,
45            preparation: settings.preparation_bytes.map(|bytes| Arc::new(Semaphore::new(bytes.get()))),
46            lanes: Arc::new(Lanes::default()),
47            registry: Arc::new(Registry {
48                middleware: crate::operations::middleware::Configuration::default(),
49                state: Mutex::new(State {
50                    closing: false,
51                    entries: BTreeMap::new(),
52                    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
53                    staging: Vec::new(),
54                    #[cfg(feature = "grpc")]
55                    grpc: BTreeMap::new(),
56                    #[cfg(feature = "grpc")]
57                    grpc_services: std::collections::BTreeSet::new(),
58                    cache: VecDeque::new(),
59                    cache_capacity: DEFAULT_CACHE_CAPACITY,
60                }),
61            }),
62        }
63    }
64
65    pub fn register_middleware(&self, middleware: Arc<dyn crate::operations::middleware::Middleware>) -> Result<()> {
66        self.registry.middleware.register(middleware)
67    }
68
69    pub fn register_authorizer(&self, authorizer: Arc<dyn crate::operations::authorization::Authorizer>) -> Result<()> {
70        self.registry.middleware.register_authorizer(authorizer)
71    }
72
73    #[cfg(feature = "grpc")]
74    pub(crate) fn submit_external_charged<O: OperationDeclaration, P: Send + 'static>(
75        &self, request: crate::operations::StreamItem<O>, endpoints: P,
76    ) -> Result<OperationInstance<O>> {
77        self.admit::<O, P>((crate::operations::binding::Request::charged(request)?, endpoints), None, &[], &Source::Host.external()?, None)
78    }
79
80    pub(in crate::operations) fn authorize_external(
81        &self, entry: &Entry, origin: &Origin,
82    ) -> Result<Option<crate::operations::authorization::Authorization>> {
83        #[cfg(feature = "grpc")]
84        if matches!(origin.identity, crate::operations::caller::Identity::External(_)) {
85            let Some(authorizer) = self.registry.middleware.authorizer() else {
86                // Preserve Existing Explicit Anonymous Routes When No Policy Is Configured.
87                return Ok(None);
88            };
89            let denied = || Error::OperationUnauthorized {
90                caller: "external".to_owned(),
91                owner: entry.owner.to_string(),
92                operation: entry.id().to_owned(),
93            };
94            let contract = entry.contract.as_ref().ok_or_else(denied)?;
95            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
96            let plugin = entry.plugin.is_some();
97            #[cfg(not(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar")))]
98            let plugin = false;
99            let selected = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
100                authorizer.authorize(
101                    self.call_context(),
102                    crate::operations::authorization::Provider {
103                        owner: &entry.owner,
104                        plugin,
105                        generation: entry.generation,
106                        definition: &entry.definition,
107                        contract,
108                    },
109                )
110            }))
111            .map_err(|_| Error::InvalidInput { field: "operation authorizer", reason: "authorization callback panicked" })??
112            .ok_or_else(denied)?;
113            let selected_contract = match &selected {
114                crate::operations::authorization::Authorization::Full(contract) => contract,
115                crate::operations::authorization::Authorization::Restricted(required) => &required.contract,
116            };
117            if !contract.compatible(selected_contract) {
118                return Err(denied());
119            }
120            return Ok(Some(selected));
121        }
122        #[cfg(not(feature = "grpc"))]
123        let _ = (entry, origin);
124        Ok(None)
125    }
126
127    pub fn with_context(&self, context: crate::operations::CallContext) -> Self {
128        let mut handle = self.clone();
129        handle.context = Arc::new(std::sync::OnceLock::from(context));
130        handle
131    }
132
133    pub(in crate::operations) fn call_context(&self) -> &crate::operations::CallContext {
134        self.context.get().expect("context freezes before operation execution")
135    }
136
137    pub fn status(&self, id: OperationInstanceId) -> Option<crate::operations::OperationStatus> {
138        self.instances.status(id)
139    }
140
141    pub(in crate::operations) fn claim_preparation(&self, bytes: std::num::NonZeroUsize) -> Result<OwnedSemaphorePermit> {
142        let count = u32::try_from(bytes.get()).map_err(|_| Error::ResourceExhausted { resource: "operation preparation bytes" })?;
143        self.preparation
144            .as_ref()
145            .ok_or(Error::InvalidInput {
146                field: "operation preparation bytes",
147                reason: "encoded preparation requires an explicit preparation budget",
148            })?
149            .clone()
150            .try_acquire_many_owned(count)
151            .map_err(|_| Error::ResourceExhausted { resource: "operation preparation bytes" })
152    }
153}
154
155/* === Public Functions === */
156
157/* === Private Functions === */
158
159/* === Tests === */
harmos_runtime/operations/manager/construction.rs

harmos_runtime/operations/manager/
construction.rs
