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

lenso_kernel/
request_handle.rs
