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

lenso_kernel/
request_handle.rs
