lenso_kernel/
request_handle.rs

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

lenso_kernel/
request_handle.rs
