1use super::*;
22
23impl 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 #[tracing::instrument(target = "harmos::operations", name = "operations.admit_at", level = "debug", skip_all, fields(operation = %id))]
92 pub(in crate::operations) fn admit_at<O: OperationDeclaration, P: Send + 'static>(
93 &self, id: &str, input: (crate::operations::binding::Request<O>, P), scope: Option<&Scope<'_>>, ancestors: &[&'static str],
94 source: &Source, cause: Option<Position>,
95 ) -> Result<OperationInstance<O>> {
96 let submitted = executor::now();
97 let origin = source.check()?;
98 let entry = self.entry_as(id, &origin)?;
99 self.admit_entry::<O, P>(input, scope, ancestors, source, Submission { submitted, entry, cause, preparation: None })
100 }
101
102 #[tracing::instrument(target = "harmos::operations", name = "operations.admit_entry", level = "debug", skip_all, fields(operation = %submission.entry.definition.id, provider = %submission.entry.owner, generation = ?submission.entry.generation, lane = submission.entry.lane, outcome = "refused"))]
103 pub(in crate::operations) fn admit_entry<O: OperationDeclaration, P: Send + 'static>(
104 &self, (mut request, endpoints): (crate::operations::binding::Request<O>, P), scope: Option<&Scope<'_>>,
105 ancestors: &[&'static str], source: &Source, submission: Submission,
106 ) -> Result<OperationInstance<O>> {
107 let Submission { submitted, entry, cause, preparation } = submission;
108 let origin = source.check()?;
109 if self.registry.state.lock().expect("operation registry lock is healthy").closing
110 && !entry.retirement_control()
111 && !(scope.is_some() && origin.has_lifecycle_loan())
112 {
113 return Err(Error::Closing { domain: "operations" });
114 }
115 entry.available_to(&origin)?;
116 entry.authorize(&origin)?;
117 let authorization = self.authorize_external(&entry, &origin)?;
118 let authority = entry.grant_authority(&origin, authorization);
119 authority.admit(O::grant(&request.value).as_ref())?;
120 let origin = Origin { grants: Some(Arc::new(authority)), ..origin };
122 source.check()?;
124 entry.available_to(&origin)?;
125 let binding = Arc::clone(&entry.binding).downcast::<Typed<O, P>>();
126 #[cfg(feature = "plugin-sidecar")]
127 let binding = binding.or_else(|original| {
128 entry.native_binding.as_ref().map_or(Err(original), |encoded| Arc::clone(encoded).downcast::<Typed<O, P>>())
129 });
130 let binding = binding.map_err(|_| Error::InvalidInput {
131 field: "operation",
132 reason: "identity belongs to a different operation or endpoint type",
133 })?;
134 let admission = self.lanes.admit(ancestors, entry.lane, entry.bounded)?;
135 let id = crate::operations::handler::fresh_instance()?;
136 let mut pending = entry.pending.lock().expect("operation settlement lock is healthy");
137 source.check()?;
138 entry.available_to(&origin)?;
139 #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
140 if entry.sdk.is_some() && pending.values().any(|control| !control.is_settled()) {
141 return Err(Error::ResourceExhausted { resource: "SDK physical settlement" });
142 }
143 let privileged = entry.lifecycle_control() || origin.has_lifecycle_loan();
144 let owner = Arc::downgrade(&entry);
145 let retired = move || {
146 if let Some(entry) = owner.upgrade() {
147 let removed = entry.pending.lock().expect("operation settlement lock is healthy").remove(&id);
148 drop(removed);
149 }
150 };
151 let reservation = match scope {
152 Some(scope) => scope.reserve_tracked(id, entry.lane, submitted, privileged, retired)?,
153 None => self.instances.reserve(id, entry.lane, submitted, privileged, retired)?,
154 };
155 let control = reservation.control();
156 control.trace().record("operation", tracing::field::display(&entry.definition.id));
157 control.trace().record("generation", tracing::field::debug(entry.generation));
158 self.instances.assign(id, origin.identity.clone());
159 pending.insert(id, Arc::clone(&control));
160 drop(pending);
161
162 let middleware = self.registry.middleware.capture();
163 let context = if middleware.is_empty() { self.context.clone() } else { Arc::new(std::sync::OnceLock::new()) };
164 request.context = context.clone();
165 let mut job = (binding.prepare)(
166 (request, endpoints),
167 id,
168 Arc::clone(&control),
169 Arc::clone(&entry),
170 admission,
171 source.clone(),
172 origin.clone(),
173 preparation,
174 )?;
175 #[cfg(feature = "grpc")]
176 if let Some(deadline) = self.call_context().get::<crate::operations::context::Deadline>() {
177 job.policy.timeout = job.policy.timeout.min(deadline.0.saturating_duration_since(submitted));
178 }
179 if !middleware.is_empty() {
180 job.middleware = Some(crate::operations::middleware::Session::new(
181 middleware,
182 entry.definition.clone(),
183 id,
184 entry.generation,
185 self.call_context().clone(),
186 context,
187 origin,
188 ));
189 }
190 let refusal = if executor::now().saturating_duration_since(submitted) >= entry.definition.policy.timeout {
191 Err(crate::Refusal::TimedOut.into())
192 } else {
193 source
194 .check()
195 .and_then(|origin| entry.available_to(&origin))
196 .and_then(|_| if self.instances.is_accepting() { Ok(()) } else { Err(Error::Closing { domain: "operation" }) })
197 };
198 if let Err(error) = refusal {
199 job.refusal = Some(error);
200 }
201 let outcome = reservation.submit(job, cause);
202 tracing::Span::current().record("outcome", "admitted");
203 Ok(OperationInstance {
204 id,
205 generation: entry.generation,
206 outcome: crate::operations::handler::Outcome::Request(outcome),
207 control,
208 released: false,
209 })
210 }
211}
212
213