lenso_kernel/
request_handle.rs

1use std::{marker::PhantomData, rc::Rc, time::Duration};
2
3use super::{
4    CancellationToken, DiagnosticAdmission, DiagnosticEvent, DiagnosticOutcome, DiagnosticSource,
5    InvocationContext, NativeAppRuntime, NativeEndpointBinding, RequestCapability, RequestId,
6    RuntimeFailure, await_with_generation_context, diagnostics::diagnostic_operation,
7    ensure_context_active, schedule_module_supervision_after_failure,
8};
9
10/// Typed, immutable native Capability endpoints materialized before App boot completes.
11#[derive(Debug)]
12pub struct NativeRequestHandle<C: RequestCapability> {
13    pub(super) endpoints: Vec<NativeEndpointBinding>,
14    pub(super) runtime: Rc<NativeAppRuntime>,
15    pub(super) caller_instance: Rc<str>,
16    pub(super) caller_is_planned: bool,
17    pub(super) allow_before_ready: bool,
18    pub(super) capability: PhantomData<fn() -> C>,
19}
20
21impl<C: RequestCapability> NativeRequestHandle<C> {
22    pub(super) fn from_endpoints(
23        endpoints: &[NativeEndpointBinding],
24        runtime: Rc<NativeAppRuntime>,
25        caller_instance: &str,
26        allow_before_ready: bool,
27    ) -> Self {
28        let caller_is_planned = runtime.plan.module_instance(caller_instance).is_some();
29        Self {
30            endpoints: endpoints.to_vec(),
31            runtime,
32            caller_instance: Rc::from(caller_instance),
33            caller_is_planned,
34            allow_before_ready,
35            capability: PhantomData,
36        }
37    }
38
39    /// Returns the number of provider endpoints captured by this handle.
40    pub fn binding_count(&self) -> usize {
41        self.endpoints.len()
42    }
43
44    fn diagnostic_caller_instance(&self) -> Option<String> {
45        self.caller_is_planned
46            .then(|| self.caller_instance.to_string())
47    }
48
49    /// Invokes a singular Capability binding without falling back across providers.
50    pub async fn invoke(
51        &self,
52        operation: &str,
53        request: C::Request,
54    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
55        let context = self.next_context();
56        self.invoke_with_context(operation, context, request).await
57    }
58
59    /// Invokes a singular binding with an explicit Invocation Context.
60    pub async fn invoke_with_context(
61        &self,
62        operation: &str,
63        context: InvocationContext,
64        request: C::Request,
65    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
66        let context = context
67            .for_caller(&self.caller_instance)
68            .for_target(C::ID, operation);
69        if let Some(endpoint) = self.endpoints.first() {
70            self.runtime
71                .diagnostics
72                .record_invocation(&self.caller_instance, &endpoint.module_instance);
73        }
74        let invocation_diagnostics = self
75            .runtime
76            .diagnostics
77            .has_interested_observer(DiagnosticSource::Invocation);
78        let started_at = invocation_diagnostics.then(|| (self.runtime.driver.now)());
79        let operation_name = invocation_diagnostics
80            .then(|| {
81                self.endpoints
82                    .first()
83                    .and_then(|endpoint| diagnostic_operation(endpoint.state.operations, operation))
84            })
85            .flatten();
86        if let Some(started_at) = started_at {
87            self.runtime
88                .diagnostics
89                .emit(DiagnosticSource::Invocation, started_at, |_| {
90                    DiagnosticEvent::InvocationStarted {
91                        request_id: context.request_id(),
92                        caller_instance: self.diagnostic_caller_instance(),
93                        provider_instance: self
94                            .endpoints
95                            .first()
96                            .map(|endpoint| endpoint.module_instance.clone()),
97                        capability: C::ID,
98                        operation: operation_name,
99                    }
100                });
101        }
102        let request_id = context.request_id();
103        let result = self
104            .invoke_with_context_inner(operation, context, request)
105            .await;
106        let outcome = request_diagnostic_outcome(&result);
107        if let Some(started_at) = started_at {
108            let completed_at = (self.runtime.driver.now)();
109            self.runtime
110                .diagnostics
111                .emit(DiagnosticSource::Invocation, completed_at, |_| {
112                    DiagnosticEvent::InvocationCompleted {
113                        request_id,
114                        caller_instance: self.diagnostic_caller_instance(),
115                        provider_instance: self
116                            .endpoints
117                            .first()
118                            .map(|endpoint| endpoint.module_instance.clone()),
119                        capability: C::ID,
120                        operation: operation_name,
121                        outcome,
122                        elapsed: completed_at.saturating_sub(started_at),
123                    }
124                });
125        }
126        if let Err(error) = &result {
127            self.runtime.diagnostics.emit_runtime_failure(
128                (self.runtime.driver.now)(),
129                self.endpoints
130                    .first()
131                    .map(|endpoint| endpoint.module_instance.as_str()),
132                error,
133            );
134            if let Some(admission) = diagnostic_admission(error) {
135                self.runtime.diagnostics.emit(
136                    DiagnosticSource::Admission,
137                    (self.runtime.driver.now)(),
138                    |_| DiagnosticEvent::AdmissionRejected {
139                        request_id,
140                        caller_instance: self.diagnostic_caller_instance(),
141                        provider_instance: self
142                            .endpoints
143                            .first()
144                            .map(|endpoint| endpoint.module_instance.clone()),
145                        capability: C::ID,
146                        operation: operation_name,
147                        outcome: admission,
148                    },
149                );
150            }
151        }
152        result
153    }
154
155    async fn invoke_with_context_inner(
156        &self,
157        operation: &str,
158        context: InvocationContext,
159        request: C::Request,
160    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
161        if self.runtime.shutdown_started.get()
162            || (!self.allow_before_ready && self.runtime.admission.is_closed())
163        {
164            return Err(RuntimeFailure::AdmissionClosed);
165        }
166        let endpoint = match self.endpoints.as_slice() {
167            [] => return Err(RuntimeFailure::Unavailable { capability: C::ID }),
168            [endpoint] => endpoint,
169            endpoints => {
170                return Err(RuntimeFailure::AmbiguousBinding {
171                    capability: C::ID,
172                    providers: endpoints.len(),
173                });
174            }
175        };
176        let snapshot = endpoint
177            .state
178            .snapshot()
179            .ok_or(RuntimeFailure::Unavailable { capability: C::ID })?;
180        let admission =
181            endpoint
182                .admission(operation)
183                .ok_or_else(|| RuntimeFailure::UnknownOperation {
184                    capability: C::ID,
185                    operation: operation.to_owned(),
186                })?;
187        let _permit = admission
188            .acquire(C::ID, operation, &context, &self.runtime.driver)
189            .await?;
190        if !endpoint.state.is_current(snapshot.generation) {
191            return Err(RuntimeFailure::Unavailable { capability: C::ID });
192        }
193        ensure_context_active(&self.runtime.driver, &context)?;
194        let outcome = await_with_generation_context(
195            &self.runtime.driver,
196            &context,
197            snapshot.cancellation,
198            C::ID,
199            C::invoke_native(
200                snapshot.endpoint.as_ref(),
201                operation,
202                request,
203                context.clone(),
204            ),
205        )
206        .await
207        .map_err(|error| {
208            schedule_module_supervision_after_failure(
209                &self.runtime,
210                &endpoint.module_instance,
211                error,
212            )
213        })?
214        .map_err(|error| {
215            schedule_module_supervision_after_failure(
216                &self.runtime,
217                &endpoint.module_instance,
218                error,
219            )
220        })?;
221        Ok(outcome)
222    }
223
224    /// Invokes every provider in the resolved many order with the same typed request.
225    pub async fn invoke_many(
226        &self,
227        operation: &str,
228        request: C::Request,
229    ) -> Result<Vec<Result<C::Response, C::DomainError>>, RuntimeFailure>
230    where
231        C::Request: Clone,
232    {
233        let context = self.next_context();
234        self.invoke_many_with_context(operation, context, request)
235            .await
236    }
237
238    /// Invokes every provider with one shared explicit Invocation Context.
239    pub async fn invoke_many_with_context(
240        &self,
241        operation: &str,
242        context: InvocationContext,
243        request: C::Request,
244    ) -> Result<Vec<Result<C::Response, C::DomainError>>, RuntimeFailure>
245    where
246        C::Request: Clone,
247    {
248        let context = context.for_caller(&self.caller_instance);
249        for endpoint in &self.endpoints {
250            self.runtime
251                .diagnostics
252                .record_invocation(&self.caller_instance, &endpoint.module_instance);
253        }
254        let invocation_diagnostics = self
255            .runtime
256            .diagnostics
257            .has_interested_observer(DiagnosticSource::Invocation);
258        let started_at = invocation_diagnostics.then(|| (self.runtime.driver.now)());
259        let operation_name = invocation_diagnostics
260            .then(|| {
261                self.endpoints
262                    .first()
263                    .and_then(|endpoint| diagnostic_operation(endpoint.state.operations, operation))
264            })
265            .flatten();
266        let request_id = context.request_id();
267        if let Some(started_at) = started_at {
268            self.runtime
269                .diagnostics
270                .emit(DiagnosticSource::Invocation, started_at, |_| {
271                    DiagnosticEvent::InvocationStarted {
272                        request_id,
273                        caller_instance: self.diagnostic_caller_instance(),
274                        provider_instance: None,
275                        capability: C::ID,
276                        operation: operation_name,
277                    }
278                });
279        }
280        let result = self
281            .invoke_many_with_context_inner(operation, context, request)
282            .await;
283        let outcome = many_request_diagnostic_outcome(&result);
284        if let Some(started_at) = started_at {
285            let completed_at = (self.runtime.driver.now)();
286            self.runtime
287                .diagnostics
288                .emit(DiagnosticSource::Invocation, completed_at, |_| {
289                    DiagnosticEvent::InvocationCompleted {
290                        request_id,
291                        caller_instance: self.diagnostic_caller_instance(),
292                        provider_instance: None,
293                        capability: C::ID,
294                        operation: operation_name,
295                        outcome,
296                        elapsed: completed_at.saturating_sub(started_at),
297                    }
298                });
299        }
300        if let Err(error) = &result {
301            self.runtime
302                .diagnostics
303                .emit_runtime_failure((self.runtime.driver.now)(), None, error);
304            if let Some(admission) = diagnostic_admission(error) {
305                self.runtime.diagnostics.emit(
306                    DiagnosticSource::Admission,
307                    (self.runtime.driver.now)(),
308                    |_| DiagnosticEvent::AdmissionRejected {
309                        request_id,
310                        caller_instance: self.diagnostic_caller_instance(),
311                        provider_instance: None,
312                        capability: C::ID,
313                        operation: operation_name,
314                        outcome: admission,
315                    },
316                );
317            }
318        }
319        result
320    }
321
322    async fn invoke_many_with_context_inner(
323        &self,
324        operation: &str,
325        context: InvocationContext,
326        request: C::Request,
327    ) -> Result<Vec<Result<C::Response, C::DomainError>>, RuntimeFailure>
328    where
329        C::Request: Clone,
330    {
331        if self.runtime.shutdown_started.get()
332            || (!self.allow_before_ready && self.runtime.admission.is_closed())
333        {
334            return Err(RuntimeFailure::AdmissionClosed);
335        }
336        if self.endpoints.is_empty() {
337            return Ok(Vec::new());
338        }
339        let mut outcomes = Vec::with_capacity(self.endpoints.len());
340        for endpoint in &self.endpoints {
341            let snapshot = endpoint
342                .state
343                .snapshot()
344                .ok_or(RuntimeFailure::Unavailable { capability: C::ID })?;
345            let admission =
346                endpoint
347                    .admission(operation)
348                    .ok_or_else(|| RuntimeFailure::UnknownOperation {
349                        capability: C::ID,
350                        operation: operation.to_owned(),
351                    })?;
352            let _permit = admission
353                .acquire(C::ID, operation, &context, &self.runtime.driver)
354                .await?;
355            if !endpoint.state.is_current(snapshot.generation) {
356                return Err(RuntimeFailure::Unavailable { capability: C::ID });
357            }
358            ensure_context_active(&self.runtime.driver, &context)?;
359            let outcome = await_with_generation_context(
360                &self.runtime.driver,
361                &context,
362                snapshot.cancellation,
363                C::ID,
364                C::invoke_native(
365                    snapshot.endpoint.as_ref(),
366                    operation,
367                    request.clone(),
368                    context.clone(),
369                ),
370            )
371            .await
372            .map_err(|error| {
373                schedule_module_supervision_after_failure(
374                    &self.runtime,
375                    &endpoint.module_instance,
376                    error,
377                )
378            })?
379            .map_err(|error| {
380                schedule_module_supervision_after_failure(
381                    &self.runtime,
382                    &endpoint.module_instance,
383                    error,
384                )
385            })?;
386            outcomes.push(outcome);
387        }
388        Ok(outcomes)
389    }
390
391    /// Creates a fresh context for a request started through this handle.
392    pub fn invocation_context(
393        &self,
394        deadline: Option<Duration>,
395        cancellation: CancellationToken,
396    ) -> InvocationContext {
397        InvocationContext::new(self.next_request_id(), deadline, cancellation)
398    }
399
400    pub(super) fn next_context(&self) -> InvocationContext {
401        self.invocation_context(None, CancellationToken::new())
402            .with_shared_caller_instance(self.caller_instance.clone())
403    }
404
405    pub(super) fn next_request_id(&self) -> RequestId {
406        let request_id = self.runtime.request_ids.get();
407        self.runtime.request_ids.set(request_id.saturating_add(1));
408        request_id
409    }
410}
411
412fn request_diagnostic_outcome<Response, DomainError>(
413    result: &Result<Result<Response, DomainError>, RuntimeFailure>,
414) -> DiagnosticOutcome {
415    match result {
416        Ok(Ok(_)) => DiagnosticOutcome::Succeeded,
417        Ok(Err(_)) => DiagnosticOutcome::DomainError,
418        Err(error) => DiagnosticOutcome::RuntimeFailure(error.into()),
419    }
420}
421
422fn many_request_diagnostic_outcome<Response, DomainError>(
423    result: &Result<Vec<Result<Response, DomainError>>, RuntimeFailure>,
424) -> DiagnosticOutcome {
425    match result {
426        Ok(outcomes) if outcomes.iter().any(Result::is_err) => DiagnosticOutcome::DomainError,
427        Ok(_) => DiagnosticOutcome::Succeeded,
428        Err(error) => DiagnosticOutcome::RuntimeFailure(error.into()),
429    }
430}
431
432fn diagnostic_admission(error: &RuntimeFailure) -> Option<DiagnosticAdmission> {
433    match error {
434        RuntimeFailure::AdmissionClosed => Some(DiagnosticAdmission::Closed),
435        RuntimeFailure::ResourceExhausted { .. } => Some(DiagnosticAdmission::Exhausted),
436        RuntimeFailure::Unavailable { .. } => Some(DiagnosticAdmission::Unavailable),
437        _ => None,
438    }
439}
lenso_kernel/request_handle.rs

lenso_kernel/
request_handle.rs
