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

lenso_kernel/
request_handle.rs
