lenso_kernel/
request_handle.rs

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

lenso_kernel/
request_handle.rs
