lenso_kernel/
request_handle.rs

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

lenso_kernel/
request_handle.rs
