harmos_runtime/operations/manager/
submission.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 fn submit<O: OperationDeclaration>(&self, request: O) -> Result<OperationInstance<O>> {
39        self.admit::<O, ()>((request.into(), ()), None, &[], &Source::Host, None)
40    }
41
42    pub fn submit_endpoints<O: OperationDeclaration, P: Send + 'static>(&self, request: O, endpoints: P) -> Result<OperationInstance<O>> {
43        self.admit::<O, P>((request.into(), endpoints), None, &[], &Source::Host, None)
44    }
45
46    pub fn submit_charged<O: OperationDeclaration, P: Send + 'static>(
47        &self, request: crate::operations::StreamItem<O>, endpoints: P,
48    ) -> Result<OperationInstance<O>> {
49        self.admit::<O, P>((crate::operations::binding::Request::charged(request)?, endpoints), None, &[], &Source::Host, None)
50    }
51
52    pub fn submit_at<O: OperationDeclaration>(&self, id: &str, request: O) -> Result<OperationInstance<O>> {
53        self.admit_at::<O, ()>(id, (request.into(), ()), None, &[], &Source::Host, None)
54    }
55
56    pub async fn call<O: OperationDeclaration>(&self, request: O) -> Result<crate::operations::Response<O::Response>> {
57        self.submit(request)?.await
58    }
59
60    pub fn submit_caused<O: OperationDeclaration>(&self, request: O, cause: Position) -> Result<OperationInstance<O>> {
61        self.admit::<O, ()>((request.into(), ()), None, &[], &Source::Host, Some(cause))
62    }
63
64    pub async fn call_caused<O: OperationDeclaration>(
65        &self, request: O, cause: Position,
66    ) -> Result<crate::operations::Response<O::Response>> {
67        self.submit_caused::<O>(request, cause)?.await
68    }
69
70    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
71    pub(in crate::operations) fn submit_by<O: OperationDeclaration, P: Send + 'static>(
72        &self, input: (crate::operations::binding::Request<O>, P), source: &Source,
73    ) -> Result<OperationInstance<O>> {
74        self.admit::<O, P>(input, None, &[], source, None)
75    }
76
77    pub(in crate::operations) fn submit_scoped<O: OperationDeclaration, P: Send + 'static>(
78        &self, input: (crate::operations::binding::Request<O>, P), scope: &Scope<'_>, admission: &Admission, source: &Source,
79        cause: Option<Position>,
80    ) -> Result<OperationInstance<O>> {
81        self.admit::<O, P>(input, Some(scope), admission.held(), source, cause)
82    }
83
84    pub(super) fn admit<O: OperationDeclaration, P: Send + 'static>(
85        &self, input: (crate::operations::binding::Request<O>, P), scope: Option<&Scope<'_>>, ancestors: &[&'static str], source: &Source,
86        cause: Option<Position>,
87    ) -> Result<OperationInstance<O>> {
88        self.admit_at::<O, P>(O::ID, input, scope, ancestors, source, cause)
89    }
90
91    pub(in crate::operations) fn admit_at<O: OperationDeclaration, P: Send + 'static>(
92        &self, id: &str, input: (crate::operations::binding::Request<O>, P), scope: Option<&Scope<'_>>, ancestors: &[&'static str],
93        source: &Source, cause: Option<Position>,
94    ) -> Result<OperationInstance<O>> {
95        let submitted = executor::now();
96        let origin = source.check()?;
97        let entry = self.entry_as(id, &origin)?;
98        self.admit_entry::<O, P>(input, scope, ancestors, source, Submission { submitted, entry, cause, preparation: None })
99    }
100
101    pub(in crate::operations) fn admit_entry<O: OperationDeclaration, P: Send + 'static>(
102        &self, (mut request, endpoints): (crate::operations::binding::Request<O>, P), scope: Option<&Scope<'_>>,
103        ancestors: &[&'static str], source: &Source, submission: Submission,
104    ) -> Result<OperationInstance<O>> {
105        let Submission { submitted, entry, cause, preparation } = submission;
106        let origin = source.check()?;
107        if self.registry.state.lock().expect("operation registry lock is healthy").closing
108            && !entry.retirement_control()
109            && !(scope.is_some() && origin.has_lifecycle_loan())
110        {
111            return Err(Error::Closing { domain: "operations" });
112        }
113        entry.available_to(&origin)?;
114        entry.authorize(&origin)?;
115        let authorization = self.authorize_external(&entry, &origin)?;
116        let authority = entry.grant_authority(&origin, authorization);
117        authority.admit(O::grant(&request.value).as_ref())?;
118        // Encoded bindings carry the authority to the provider's prepared grant.
119        let origin = Origin { grants: Some(Arc::new(authority)), ..origin };
120        // Authoring Callbacks Run Outside Registry/Pending Locks. Recheck The Provider.
121        source.check()?;
122        entry.available_to(&origin)?;
123        let binding = Arc::clone(&entry.binding).downcast::<Typed<O, P>>();
124        #[cfg(feature = "plugin-sidecar")]
125        let binding = binding.or_else(|original| {
126            entry.native_binding.as_ref().map_or(Err(original), |encoded| Arc::clone(encoded).downcast::<Typed<O, P>>())
127        });
128        let binding = binding.map_err(|_| Error::InvalidInput {
129            field: "operation",
130            reason: "identity belongs to a different operation or endpoint type",
131        })?;
132        let admission = self.lanes.admit(ancestors, entry.lane, entry.bounded)?;
133        let id = crate::operations::handler::fresh_instance()?;
134        let mut pending = entry.pending.lock().expect("operation settlement lock is healthy");
135        source.check()?;
136        entry.available_to(&origin)?;
137        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
138        if entry.sdk.is_some() && pending.values().any(|control| !control.is_settled()) {
139            return Err(Error::ResourceExhausted { resource: "SDK physical settlement" });
140        }
141        let privileged = entry.lifecycle_control() || origin.has_lifecycle_loan();
142        let owner = Arc::downgrade(&entry);
143        let retired = move || {
144            if let Some(entry) = owner.upgrade() {
145                let removed = entry.pending.lock().expect("operation settlement lock is healthy").remove(&id);
146                drop(removed);
147            }
148        };
149        let reservation = match scope {
150            Some(scope) => scope.reserve_tracked(id, entry.lane, submitted, privileged, retired)?,
151            None => self.instances.reserve(id, entry.lane, submitted, privileged, retired)?,
152        };
153        let control = reservation.control();
154        self.instances.assign(id, origin.identity.clone());
155        pending.insert(id, Arc::clone(&control));
156        drop(pending);
157
158        let middleware = self.registry.middleware.capture();
159        let context = if middleware.is_empty() { self.context.clone() } else { Arc::new(std::sync::OnceLock::new()) };
160        request.context = context.clone();
161        let mut job = (binding.prepare)(
162            (request, endpoints),
163            id,
164            Arc::clone(&control),
165            Arc::clone(&entry),
166            admission,
167            source.clone(),
168            origin.clone(),
169            preparation,
170        )?;
171        #[cfg(feature = "grpc")]
172        if let Some(deadline) = self.call_context().get::<crate::operations::context::Deadline>() {
173            job.policy.timeout = job.policy.timeout.min(deadline.0.saturating_duration_since(submitted));
174        }
175        if !middleware.is_empty() {
176            job.middleware = Some(crate::operations::middleware::Session::new(
177                middleware,
178                entry.definition.clone(),
179                id,
180                entry.generation,
181                self.call_context().clone(),
182                context,
183                origin,
184            ));
185        }
186        let refusal = if executor::now().saturating_duration_since(submitted) >= entry.definition.policy.timeout {
187            Err(crate::Refusal::TimedOut.into())
188        } else {
189            source
190                .check()
191                .and_then(|origin| entry.available_to(&origin))
192                .and_then(|_| if self.instances.is_accepting() { Ok(()) } else { Err(Error::Closing { domain: "operation" }) })
193        };
194        if let Err(error) = refusal {
195            job.refusal = Some(error);
196        }
197        let outcome = reservation.submit(job, cause);
198        Ok(OperationInstance {
199            id,
200            generation: entry.generation,
201            outcome: crate::operations::handler::Outcome::Request(outcome),
202            control,
203            released: false,
204        })
205    }
206}
207
208/* === Public Functions === */
209
210/* === Private Functions === */
211
212/* === Tests === */
harmos_runtime/operations/manager/submission.rs

harmos_runtime/operations/manager/
submission.rs
