lenso_kernel/
runtime.rs

1use super::{
2    AppAdmission, AppReadyGate, BTreeMap, CancellationToken, Cell, DiagnosticEvent,
3    DiagnosticShutdownOutcome, DiagnosticSource, DriverControl, DriverTask, Duration,
4    EventCapability, ExecutionAdapterCatalog, InvocationContext, LocalBoxFuture,
5    ManagedResourceScope, ManagedTask, ManagedTaskScope, ModuleCriticality, ModuleDependencies,
6    ModuleLifecycle, NativeEventBindingTable, NativeEventEndpointStateTable, NativeEventHandle,
7    NativeRequestEndpoint, NativeRequestHandle, NativeStreamBindingTable, NativeStreamEndpoint,
8    NativeStreamEndpointStateTable, NativeStreamHandle, Rc, RefCell, RequestAdmission,
9    RequestCapability, RequestId, ResolvedAppPlan, RestartPolicy, RuntimeDiagnostics,
10    RuntimeFailure, ShutdownOutcome, StreamCapability, begin_module_supervision, event,
11    handle_supervision_schedule_failure, oneshot, schedule_module_supervision,
12    shutdown_native_modules,
13};
14
15#[derive(Clone, Debug)]
16pub(super) struct NativeEndpointSnapshot {
17    pub(super) endpoint: Rc<dyn NativeRequestEndpoint>,
18    pub(super) generation: u64,
19    pub(super) cancellation: CancellationToken,
20}
21
22#[derive(Debug)]
23pub(super) struct NativeEndpointState {
24    pub(super) capability_id: &'static str,
25    pub(super) descriptor_version: &'static str,
26    pub(super) operations: &'static [&'static str],
27    pub(super) endpoint: RefCell<Option<Rc<dyn NativeRequestEndpoint>>>,
28    pub(super) generation: Cell<u64>,
29    pub(super) cancellation: RefCell<CancellationToken>,
30}
31
32#[derive(Clone, Debug)]
33pub(crate) struct NativeStreamEndpointSnapshot {
34    pub(crate) endpoint: Rc<dyn NativeStreamEndpoint>,
35    pub(crate) generation: u64,
36    pub(crate) cancellation: CancellationToken,
37}
38
39#[derive(Debug)]
40pub(crate) struct NativeStreamEndpointState {
41    pub(super) capability_id: &'static str,
42    pub(super) descriptor_version: &'static str,
43    pub(super) operations: &'static [&'static str],
44    pub(super) endpoint: RefCell<Option<Rc<dyn NativeStreamEndpoint>>>,
45    pub(super) generation: Cell<u64>,
46    pub(super) cancellation: RefCell<CancellationToken>,
47}
48
49impl NativeStreamEndpointState {
50    pub(crate) fn new(endpoint: Rc<dyn NativeStreamEndpoint>, generation: u64) -> Self {
51        Self {
52            capability_id: endpoint.capability_id(),
53            descriptor_version: endpoint.descriptor_version(),
54            operations: endpoint.operations(),
55            endpoint: RefCell::new(Some(endpoint)),
56            generation: Cell::new(generation),
57            cancellation: RefCell::new(CancellationToken::new()),
58        }
59    }
60
61    pub(crate) fn snapshot(&self) -> Option<NativeStreamEndpointSnapshot> {
62        self.endpoint
63            .borrow()
64            .clone()
65            .map(|endpoint| NativeStreamEndpointSnapshot {
66                endpoint,
67                generation: self.generation.get(),
68                cancellation: self.cancellation.borrow().clone(),
69            })
70    }
71
72    pub(crate) fn mark_unavailable(&self) {
73        self.cancellation.borrow().cancel();
74        self.endpoint.borrow_mut().take();
75    }
76
77    pub(crate) fn install(&self, endpoint: Rc<dyn NativeStreamEndpoint>, generation: u64) {
78        self.generation.set(generation);
79        self.cancellation.replace(CancellationToken::new());
80        self.endpoint.replace(Some(endpoint));
81    }
82
83    pub(crate) fn is_current(&self, generation: u64) -> bool {
84        self.generation.get() == generation && self.endpoint.borrow().is_some()
85    }
86}
87
88impl NativeEndpointState {
89    pub(super) fn new(endpoint: Rc<dyn NativeRequestEndpoint>, generation: u64) -> Self {
90        Self {
91            capability_id: endpoint.capability_id(),
92            descriptor_version: endpoint.descriptor_version(),
93            operations: endpoint.operations(),
94            endpoint: RefCell::new(Some(endpoint)),
95            generation: Cell::new(generation),
96            cancellation: RefCell::new(CancellationToken::new()),
97        }
98    }
99
100    pub(super) fn snapshot(&self) -> Option<NativeEndpointSnapshot> {
101        self.endpoint
102            .borrow()
103            .clone()
104            .map(|endpoint| NativeEndpointSnapshot {
105                endpoint,
106                generation: self.generation.get(),
107                cancellation: self.cancellation.borrow().clone(),
108            })
109    }
110
111    pub(super) fn mark_unavailable(&self) {
112        self.cancellation.borrow().cancel();
113        self.endpoint.borrow_mut().take();
114    }
115
116    pub(super) fn install(&self, endpoint: Rc<dyn NativeRequestEndpoint>, generation: u64) {
117        self.generation.set(generation);
118        self.cancellation.replace(CancellationToken::new());
119        self.endpoint.replace(Some(endpoint));
120    }
121
122    pub(super) fn is_current(&self, generation: u64) -> bool {
123        self.generation.get() == generation && self.endpoint.borrow().is_some()
124    }
125}
126
127#[derive(Clone, Debug)]
128pub(super) struct NativeEndpointBinding {
129    pub(super) module_instance: String,
130    pub(super) state: Rc<NativeEndpointState>,
131    pub(super) admissions: BTreeMap<String, RequestAdmission>,
132}
133
134impl NativeEndpointBinding {
135    pub(super) fn admission(&self, operation: &str) -> Option<&RequestAdmission> {
136        self.admissions.get(operation)
137    }
138}
139
140#[derive(Clone, Debug)]
141pub(crate) struct NativeStreamEndpointBinding {
142    pub(crate) module_instance: String,
143    pub(crate) state: Rc<NativeStreamEndpointState>,
144    pub(super) admissions: BTreeMap<String, RequestAdmission>,
145}
146
147impl NativeStreamEndpointBinding {
148    pub(crate) fn admission(&self, operation: &str) -> Option<&RequestAdmission> {
149        self.admissions.get(operation)
150    }
151}
152
153#[derive(Debug)]
154pub(super) struct NativeModuleGeneration {
155    pub(super) lifecycle: Rc<dyn ModuleLifecycle>,
156    pub(super) tasks: ManagedTaskScope,
157    pub(super) resources: ManagedResourceScope,
158}
159
160pub(super) enum GenerationPreparationFailure {
161    Lifecycle,
162    Cleanup(RuntimeFailure),
163}
164
165#[derive(Debug)]
166pub(super) struct NativeModuleRuntime {
167    pub(super) generation: RefCell<Option<NativeModuleGeneration>>,
168}
169
170impl NativeModuleRuntime {
171    pub(super) fn take_generation(&self) -> Option<NativeModuleGeneration> {
172        self.generation.borrow_mut().take()
173    }
174
175    pub(super) fn install_generation(&self, generation: NativeModuleGeneration) {
176        debug_assert!(self.generation.borrow().is_none());
177        self.generation.replace(Some(generation));
178    }
179
180    pub(super) fn generation_parts(
181        &self,
182    ) -> Option<(
183        Rc<dyn ModuleLifecycle>,
184        ManagedTaskScope,
185        ManagedResourceScope,
186    )> {
187        self.generation.borrow().as_ref().map(|generation| {
188            (
189                generation.lifecycle.clone(),
190                generation.tasks.clone(),
191                generation.resources.clone(),
192            )
193        })
194    }
195}
196
197#[derive(Clone, Debug)]
198pub(super) struct ModuleSupervision {
199    pub(super) policy: RestartPolicy,
200    pub(super) criticality: ModuleCriticality,
201    pub(super) required_path: bool,
202    pub(super) generation: u64,
203    pub(super) attempts: Vec<Duration>,
204    pub(super) stable_since: Option<Duration>,
205    pub(super) restarting: bool,
206}
207
208#[derive(Debug, Default)]
209pub(super) struct ShutdownCoordinator {
210    pub(super) started: Cell<bool>,
211    pub(super) cleanup_started_at: Cell<Option<Duration>>,
212    pub(super) completed: Cell<bool>,
213    pub(super) outcome: RefCell<Option<ShutdownOutcome>>,
214    pub(super) waiters: RefCell<Vec<oneshot::Sender<ShutdownOutcome>>>,
215}
216
217impl ShutdownCoordinator {
218    pub(super) fn start(&self, started_at: Duration) -> bool {
219        if self.started.replace(true) {
220            return false;
221        }
222        self.cleanup_started_at.set(Some(started_at));
223        true
224    }
225
226    pub(super) fn begin_completion(&self) -> bool {
227        !self.completed.replace(true)
228    }
229
230    pub(super) fn publish(&self, outcome: &ShutdownOutcome) {
231        self.outcome.replace(Some(outcome.clone()));
232        for waiter in self.waiters.borrow_mut().drain(..) {
233            let _ = waiter.send(outcome.clone());
234        }
235    }
236
237    pub(super) fn wait(&self) -> LocalBoxFuture<'static, ShutdownOutcome> {
238        if let Some(outcome) = self.outcome.borrow().clone() {
239            return Box::pin(futures::future::ready(outcome));
240        }
241        let (complete, waiter) = oneshot::channel();
242        self.waiters.borrow_mut().push(complete);
243        Box::pin(async move {
244            waiter.await.unwrap_or(ShutdownOutcome::RuntimeFailure {
245                error: RuntimeFailure::Internal {
246                    detail: "shutdown coordinator terminated before publishing an outcome"
247                        .to_owned(),
248                },
249            })
250        })
251    }
252}
253
254pub(super) struct NativeAppRuntime {
255    pub(super) plan: ResolvedAppPlan,
256    pub(super) adapters: Rc<ExecutionAdapterCatalog>,
257    pub(super) modules: BTreeMap<String, NativeModuleRuntime>,
258    pub(super) dependencies: BTreeMap<String, ModuleDependencies>,
259    pub(super) endpoint_states: BTreeMap<(String, String), Rc<NativeEndpointState>>,
260    pub(super) stream_endpoint_states: NativeStreamEndpointStateTable,
261    pub(super) event_endpoint_states: NativeEventEndpointStateTable,
262    pub(super) supervision: RefCell<BTreeMap<String, ModuleSupervision>>,
263    pub(super) supervision_tasks: RefCell<BTreeMap<String, ManagedTask>>,
264    pub(super) activation_order: Vec<String>,
265    pub(super) ready_gate: AppReadyGate,
266    pub(super) admission: AppAdmission,
267    pub(super) driver: DriverControl,
268    pub(super) diagnostics: RuntimeDiagnostics,
269    pub(super) request_ids: Rc<Cell<RequestId>>,
270    pub(super) supervision_cancellation: CancellationToken,
271    pub(super) shutdown_started: Cell<bool>,
272    pub(super) shutdown: ShutdownCoordinator,
273    pub(super) shutdown_task: RefCell<Option<DriverTask>>,
274    pub(super) terminal_failure: RefCell<Option<RuntimeFailure>>,
275}
276
277impl std::fmt::Debug for NativeAppRuntime {
278    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
279        formatter
280            .debug_struct("NativeAppRuntime")
281            .field("module_count", &self.modules.len())
282            .field("endpoint_count", &self.endpoint_states.len())
283            .field("stream_endpoint_count", &self.stream_endpoint_states.len())
284            .field("event_endpoint_count", &self.event_endpoint_states.len())
285            .field("ready", &self.ready_gate.is_open())
286            .field("accepting", &self.admission.is_open())
287            .field("next_request_id", &self.request_ids.get())
288            .field("shutdown_started", &self.shutdown_started.get())
289            .field("cleanup_started", &self.shutdown.started.get())
290            .field("cleanup_completed", &self.shutdown.completed.get())
291            .field(
292                "terminal_failure",
293                &self.terminal_failure.borrow().is_some(),
294            )
295            .finish_non_exhaustive()
296    }
297}
298
299impl NativeAppRuntime {
300    pub(super) fn begin_shutdown(&self) {
301        let admission_closed_at = (self.driver.now)();
302        if self.shutdown_started.replace(true) {
303            return;
304        }
305        self.admission.close();
306        self.supervision_cancellation.cancel();
307        for endpoint in self.endpoint_states.values() {
308            endpoint.mark_unavailable();
309        }
310        for endpoint in self.stream_endpoint_states.values() {
311            endpoint.mark_unavailable();
312        }
313        for endpoint in self.event_endpoint_states.values() {
314            endpoint.mark_unavailable();
315        }
316        for module in self.modules.values() {
317            if let Some((_, tasks, resources)) = module.generation_parts() {
318                tasks.close();
319                resources.close();
320            }
321        }
322        self.diagnostics
323            .emit(DiagnosticSource::Shutdown, admission_closed_at, |_| {
324                DiagnosticEvent::ShutdownAdmissionClosed
325            });
326    }
327
328    pub(super) fn complete_shutdown(&self, outcome: &ShutdownOutcome) {
329        if !self.shutdown.begin_completion() {
330            return;
331        }
332        let completed_at = (self.driver.now)();
333        let started_at = self
334            .shutdown
335            .cleanup_started_at
336            .get()
337            .unwrap_or(completed_at);
338        let diagnostic_outcome = match outcome {
339            ShutdownOutcome::Clean => DiagnosticShutdownOutcome::Clean,
340            ShutdownOutcome::RuntimeFailure { .. } => DiagnosticShutdownOutcome::RuntimeFailure,
341            ShutdownOutcome::Timeout => DiagnosticShutdownOutcome::Timeout,
342        };
343        self.diagnostics
344            .emit(DiagnosticSource::Shutdown, completed_at, |_| {
345                DiagnosticEvent::ShutdownCompleted {
346                    outcome: diagnostic_outcome,
347                    elapsed: completed_at.saturating_sub(started_at),
348                }
349            });
350        if let ShutdownOutcome::RuntimeFailure { error } = outcome {
351            self.diagnostics
352                .emit_runtime_failure(completed_at, None, error);
353        }
354        self.shutdown.publish(outcome);
355    }
356}
357
358/// A started native App whose generated clients can invoke resolved bindings.
359#[derive(Clone, Debug)]
360pub struct NativeApp {
361    pub(super) bindings: BTreeMap<(String, &'static str), Vec<NativeEndpointBinding>>,
362    pub(super) stream_bindings: NativeStreamBindingTable,
363    pub(super) event_bindings: NativeEventBindingTable,
364    pub(super) diagnostics: RuntimeDiagnostics,
365    pub(super) runtime: Rc<NativeAppRuntime>,
366}
367
368impl NativeApp {
369    fn diagnostic_failure<T>(
370        &self,
371        instance_key: Option<&str>,
372        error: RuntimeFailure,
373    ) -> Result<T, RuntimeFailure> {
374        let instance_key = instance_key
375            .filter(|instance_key| self.runtime.plan.module_instance(instance_key).is_some());
376        self.runtime.diagnostics.emit_runtime_failure(
377            (self.runtime.driver.now)(),
378            instance_key,
379            &error,
380        );
381        Err(error)
382    }
383
384    /// Confirms that a generated client has one resolved binding before use.
385    pub fn ensure_binding<C: RequestCapability>(
386        &self,
387        caller_instance: &str,
388    ) -> Result<(), RuntimeFailure> {
389        if self.runtime.admission.is_closed() {
390            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
391        }
392        if self
393            .endpoints::<C>(caller_instance)
394            .is_some_and(|endpoints| !endpoints.is_empty())
395        {
396            return Ok(());
397        }
398        self.diagnostic_failure(
399            Some(caller_instance),
400            RuntimeFailure::Unavailable { capability: C::ID },
401        )
402    }
403
404    /// Materializes one typed handle from the immutable binding selected by the Plan.
405    pub fn handle<C: RequestCapability>(
406        &self,
407        caller_instance: &str,
408    ) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
409        if self.runtime.admission.is_closed() {
410            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
411        }
412        let Some(endpoints) = self
413            .endpoints::<C>(caller_instance)
414            .filter(|endpoints| !endpoints.is_empty())
415        else {
416            return self.diagnostic_failure(
417                Some(caller_instance),
418                RuntimeFailure::Unavailable { capability: C::ID },
419            );
420        };
421        Ok(NativeRequestHandle::from_endpoints(
422            endpoints,
423            self.runtime.clone(),
424            caller_instance,
425            false,
426        ))
427    }
428
429    /// Materializes an optional typed handle; an absent binding remains `None`.
430    pub fn optional_handle<C: RequestCapability>(
431        &self,
432        caller_instance: &str,
433    ) -> Option<NativeRequestHandle<C>> {
434        let caller_instance = caller_instance.to_owned();
435        self.endpoints::<C>(&caller_instance)
436            .filter(|endpoints| !endpoints.is_empty())
437            .map(|endpoints| {
438                NativeRequestHandle::from_endpoints(
439                    endpoints,
440                    self.runtime.clone(),
441                    &caller_instance,
442                    false,
443                )
444            })
445    }
446
447    /// Materializes a typed handle whose endpoints may be empty for a `many` requirement.
448    pub fn many_handle<C: RequestCapability>(
449        &self,
450        caller_instance: &str,
451    ) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
452        if self.runtime.admission.is_closed() {
453            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
454        }
455        let endpoints = self.endpoints::<C>(caller_instance).unwrap_or(&[]);
456        Ok(NativeRequestHandle::from_endpoints(
457            endpoints,
458            self.runtime.clone(),
459            caller_instance,
460            false,
461        ))
462    }
463
464    /// Returns the number of immutable provider endpoints bound to one requirement.
465    pub fn binding_count<C: RequestCapability>(&self, caller_instance: &str) -> usize {
466        self.endpoints::<C>(caller_instance).map_or(0, <[_]>::len)
467    }
468
469    /// Returns whether every declared Module has completed activation.
470    pub fn is_ready(&self) -> bool {
471        self.runtime.ready_gate.is_open()
472    }
473
474    /// Returns the App-wide readiness signal observed by Module tasks.
475    pub fn ready_gate(&self) -> AppReadyGate {
476        self.runtime.ready_gate.clone()
477    }
478
479    /// Returns whether new externally triggered work may be admitted.
480    pub fn is_accepting(&self) -> bool {
481        self.runtime.admission.is_open()
482    }
483
484    /// Returns the App-wide admission state.
485    pub fn admission(&self) -> AppAdmission {
486        self.runtime.admission.clone()
487    }
488
489    /// Returns the opt-in Runtime Diagnostics port for this App.
490    pub fn diagnostics(&self) -> RuntimeDiagnostics {
491        self.diagnostics.clone()
492    }
493
494    /// Returns the exact resolved Capability dependencies for one Module Instance.
495    ///
496    /// Runners use this host-facing view when they must preserve provider identity
497    /// while transferring one binding across an Execution Lane. Ordinary Module
498    /// code receives the same Interface through its lifecycle context.
499    pub fn dependencies(
500        &self,
501        caller_instance: &str,
502    ) -> Result<ModuleDependencies, RuntimeFailure> {
503        if self.runtime.admission.is_closed() {
504            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
505        }
506        self.runtime
507            .dependencies
508            .get(caller_instance)
509            .cloned()
510            .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
511                detail: format!(
512                    "Module Instance `{caller_instance}` has no resolved dependency table"
513                ),
514            })
515    }
516
517    /// Returns current bounded request queue depths grouped by provider Instance.
518    ///
519    /// This structural snapshot contains no request payloads and is intended for
520    /// Runner-owned placement diagnostics.
521    pub fn instance_queue_depths(&self) -> BTreeMap<String, usize> {
522        let mut depths = BTreeMap::new();
523        for endpoints in self.bindings.values() {
524            for endpoint in endpoints {
525                let depth = endpoint
526                    .admissions
527                    .values()
528                    .map(RequestAdmission::queue_depth)
529                    .sum::<usize>();
530                *depths.entry(endpoint.module_instance.clone()).or_insert(0) += depth;
531            }
532        }
533        depths
534    }
535
536    /// Returns the terminal supervision failure, when a critical App path exhausted its budget.
537    pub fn terminal_failure(&self) -> Option<RuntimeFailure> {
538        self.runtime.terminal_failure.borrow().clone()
539    }
540
541    /// Returns whether supervision has produced a terminal App failure.
542    pub fn is_failed(&self) -> bool {
543        self.runtime.terminal_failure.borrow().is_some()
544    }
545
546    /// Returns the current ready generation for one Module Instance, when it is available.
547    pub fn module_generation(&self, instance_key: &str) -> Option<u64> {
548        self.runtime
549            .supervision
550            .borrow()
551            .get(instance_key)
552            .and_then(|state| {
553                let request_current =
554                    self.runtime
555                        .endpoint_states
556                        .iter()
557                        .any(|((module, _), endpoint)| {
558                            module == instance_key && endpoint.is_current(state.generation)
559                        });
560                let stream_current =
561                    self.runtime
562                        .stream_endpoint_states
563                        .iter()
564                        .any(|((module, _), endpoint)| {
565                            module == instance_key && endpoint.is_current(state.generation)
566                        });
567                let event_current =
568                    self.runtime
569                        .event_endpoint_states
570                        .iter()
571                        .any(|((module, _), endpoint)| {
572                            module == instance_key && endpoint.is_current(state.generation)
573                        });
574                (request_current || stream_current || event_current).then_some(state.generation)
575            })
576    }
577
578    /// Reports a Module Instance failure and schedules its finite supervision policy.
579    pub fn report_module_failure(&self, instance_key: &str) -> Result<(), RuntimeFailure> {
580        if !begin_module_supervision(&self.runtime, instance_key)? {
581            return Ok(());
582        }
583        schedule_module_supervision(&self.runtime, instance_key).map_err(|error| {
584            handle_supervision_schedule_failure(&self.runtime, instance_key, error)
585        })
586    }
587
588    /// Starts shutdown admission closure and cooperative cancellation.
589    pub fn request_shutdown(&self) {
590        self.runtime.begin_shutdown();
591    }
592
593    /// Performs bounded graceful shutdown using one global deadline.
594    pub async fn shutdown(&self, timeout: Duration) -> ShutdownOutcome {
595        self.runtime.begin_shutdown();
596        let cleanup_started_at = (self.runtime.driver.now)();
597        if self.runtime.shutdown.start(cleanup_started_at) {
598            self.runtime
599                .diagnostics
600                .emit(DiagnosticSource::Shutdown, cleanup_started_at, |_| {
601                    DiagnosticEvent::ShutdownCleanupStarted { timeout }
602                });
603            let runtime = self.runtime.clone();
604            let worker_runtime = runtime.clone();
605            match (runtime.driver.spawn_local)(Box::pin(async move {
606                let outcome = shutdown_native_modules(&worker_runtime, timeout).await;
607                worker_runtime.complete_shutdown(&outcome);
608            })) {
609                Ok(task) => {
610                    runtime.shutdown_task.replace(Some(task));
611                }
612                Err(error) => {
613                    runtime.complete_shutdown(&ShutdownOutcome::RuntimeFailure {
614                        error: RuntimeFailure::Internal {
615                            detail: format!("failed to schedule App shutdown: {error:?}"),
616                        },
617                    });
618                }
619            }
620        }
621        self.runtime.shutdown.wait().await
622    }
623
624    /// Invokes a generated request Operation through the caller's resolved binding.
625    pub async fn invoke<C: RequestCapability>(
626        &self,
627        caller_instance: &str,
628        operation: &str,
629        request: C::Request,
630    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
631        self.handle::<C>(caller_instance)?
632            .invoke(operation, request)
633            .await
634    }
635
636    /// Creates a request context with a fresh Kernel Request ID.
637    ///
638    /// `deadline` is an absolute instant returned by the selected
639    /// [`RuntimeDriver`]'s monotonic clock.
640    pub fn invocation_context(
641        &self,
642        deadline: Option<Duration>,
643        cancellation: CancellationToken,
644    ) -> InvocationContext {
645        InvocationContext::new(self.next_request_id(), deadline, cancellation)
646    }
647
648    /// Creates a request context whose deadline is relative to the Driver's clock.
649    pub fn invocation_context_after(
650        &self,
651        timeout: Duration,
652        cancellation: CancellationToken,
653    ) -> InvocationContext {
654        self.invocation_context(
655            Some((self.runtime.driver.now)().saturating_add(timeout)),
656            cancellation,
657        )
658    }
659
660    /// Invokes a request with an explicit propagated Invocation Context.
661    pub async fn invoke_with_context<C: RequestCapability>(
662        &self,
663        caller_instance: &str,
664        operation: &str,
665        context: InvocationContext,
666        request: C::Request,
667    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
668        self.handle::<C>(caller_instance)?
669            .invoke_with_context(operation, context, request)
670            .await
671    }
672
673    /// Materializes one typed bidirectional stream handle from the resolved Plan.
674    pub fn stream_handle<C: StreamCapability>(
675        &self,
676        caller_instance: &str,
677    ) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
678        if self.runtime.admission.is_closed() {
679            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
680        }
681        let Some(endpoints) = self
682            .stream_endpoints::<C>(caller_instance)
683            .filter(|endpoints| !endpoints.is_empty())
684        else {
685            return self.diagnostic_failure(
686                Some(caller_instance),
687                RuntimeFailure::Unavailable { capability: C::ID },
688            );
689        };
690        Ok(NativeStreamHandle::from_endpoints(
691            endpoints,
692            self.runtime.clone(),
693            caller_instance,
694            false,
695        ))
696    }
697
698    /// Materializes an optional typed bidirectional stream handle.
699    pub fn optional_stream_handle<C: StreamCapability>(
700        &self,
701        caller_instance: &str,
702    ) -> Option<NativeStreamHandle<C>> {
703        let caller_instance = caller_instance.to_owned();
704        self.stream_endpoints::<C>(&caller_instance)
705            .filter(|endpoints| !endpoints.is_empty())
706            .map(|endpoints| {
707                NativeStreamHandle::from_endpoints(
708                    endpoints,
709                    self.runtime.clone(),
710                    &caller_instance,
711                    false,
712                )
713            })
714    }
715
716    /// Returns the number of immutable stream endpoints bound to one requirement.
717    pub fn stream_binding_count<C: StreamCapability>(&self, caller_instance: &str) -> usize {
718        self.stream_endpoints::<C>(caller_instance)
719            .map_or(0, <[_]>::len)
720    }
721
722    /// Materializes a typed Event handle and requires at least one subscriber.
723    pub fn event_handle<C: EventCapability>(
724        &self,
725        caller_instance: &str,
726    ) -> Result<NativeEventHandle<C>, RuntimeFailure> {
727        if self.runtime.admission.is_closed() {
728            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
729        }
730        let Some(endpoints) = self
731            .event_endpoints::<C>(caller_instance)
732            .filter(|endpoints| !endpoints.is_empty())
733        else {
734            return self.diagnostic_failure(
735                Some(caller_instance),
736                RuntimeFailure::Unavailable { capability: C::ID },
737            );
738        };
739        Ok(NativeEventHandle::from_endpoints(
740            endpoints,
741            self.runtime.clone(),
742            caller_instance,
743            false,
744        ))
745    }
746
747    /// Materializes an optional typed Event handle.
748    pub fn optional_event_handle<C: EventCapability>(
749        &self,
750        caller_instance: &str,
751    ) -> Option<NativeEventHandle<C>> {
752        let caller_instance = caller_instance.to_owned();
753        self.event_endpoints::<C>(&caller_instance)
754            .filter(|endpoints| !endpoints.is_empty())
755            .map(|endpoints| {
756                NativeEventHandle::from_endpoints(
757                    endpoints,
758                    self.runtime.clone(),
759                    &caller_instance,
760                    false,
761                )
762            })
763    }
764
765    /// Materializes a typed Event handle whose endpoint set may be empty.
766    pub fn many_event_handle<C: EventCapability>(
767        &self,
768        caller_instance: &str,
769    ) -> Result<NativeEventHandle<C>, RuntimeFailure> {
770        if self.runtime.admission.is_closed() {
771            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
772        }
773        let endpoints = self.event_endpoints::<C>(caller_instance).unwrap_or(&[]);
774        Ok(NativeEventHandle::from_endpoints(
775            endpoints,
776            self.runtime.clone(),
777            caller_instance,
778            false,
779        ))
780    }
781
782    /// Returns the number of immutable Event subscriber endpoints bound to a requirement.
783    pub fn event_binding_count<C: EventCapability>(&self, caller_instance: &str) -> usize {
784        self.event_endpoints::<C>(caller_instance)
785            .map_or(0, <[_]>::len)
786    }
787
788    pub(super) fn next_request_id(&self) -> RequestId {
789        let request_id = self.runtime.request_ids.get();
790        self.runtime.request_ids.set(request_id.saturating_add(1));
791        request_id
792    }
793
794    pub(super) fn endpoints<C: RequestCapability>(
795        &self,
796        caller_instance: &str,
797    ) -> Option<&[NativeEndpointBinding]> {
798        self.bindings
799            .get(&(caller_instance.to_owned(), C::ID))
800            .map(Vec::as_slice)
801    }
802
803    pub(super) fn stream_endpoints<C: StreamCapability>(
804        &self,
805        caller_instance: &str,
806    ) -> Option<&[NativeStreamEndpointBinding]> {
807        self.stream_bindings
808            .get(&(caller_instance.to_owned(), C::ID))
809            .map(Vec::as_slice)
810    }
811
812    pub(super) fn event_endpoints<C: EventCapability>(
813        &self,
814        caller_instance: &str,
815    ) -> Option<&[event::NativeEventEndpointBinding]> {
816        self.event_bindings
817            .get(&(caller_instance.to_owned(), C::ID))
818            .map(Vec::as_slice)
819    }
820}
lenso_kernel/runtime.rs

lenso_kernel/
runtime.rs
