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 current bounded request queue depths grouped by provider Instance.
495    ///
496    /// This structural snapshot contains no request payloads and is intended for
497    /// Runner-owned placement diagnostics.
498    pub fn instance_queue_depths(&self) -> BTreeMap<String, usize> {
499        let mut depths = BTreeMap::new();
500        for endpoints in self.bindings.values() {
501            for endpoint in endpoints {
502                let depth = endpoint
503                    .admissions
504                    .values()
505                    .map(RequestAdmission::queue_depth)
506                    .sum::<usize>();
507                *depths.entry(endpoint.module_instance.clone()).or_insert(0) += depth;
508            }
509        }
510        depths
511    }
512
513    /// Returns the terminal supervision failure, when a critical App path exhausted its budget.
514    pub fn terminal_failure(&self) -> Option<RuntimeFailure> {
515        self.runtime.terminal_failure.borrow().clone()
516    }
517
518    /// Returns whether supervision has produced a terminal App failure.
519    pub fn is_failed(&self) -> bool {
520        self.runtime.terminal_failure.borrow().is_some()
521    }
522
523    /// Returns the current ready generation for one Module Instance, when it is available.
524    pub fn module_generation(&self, instance_key: &str) -> Option<u64> {
525        self.runtime
526            .supervision
527            .borrow()
528            .get(instance_key)
529            .and_then(|state| {
530                let request_current =
531                    self.runtime
532                        .endpoint_states
533                        .iter()
534                        .any(|((module, _), endpoint)| {
535                            module == instance_key && endpoint.is_current(state.generation)
536                        });
537                let stream_current =
538                    self.runtime
539                        .stream_endpoint_states
540                        .iter()
541                        .any(|((module, _), endpoint)| {
542                            module == instance_key && endpoint.is_current(state.generation)
543                        });
544                let event_current =
545                    self.runtime
546                        .event_endpoint_states
547                        .iter()
548                        .any(|((module, _), endpoint)| {
549                            module == instance_key && endpoint.is_current(state.generation)
550                        });
551                (request_current || stream_current || event_current).then_some(state.generation)
552            })
553    }
554
555    /// Reports a Module Instance failure and schedules its finite supervision policy.
556    pub fn report_module_failure(&self, instance_key: &str) -> Result<(), RuntimeFailure> {
557        if !begin_module_supervision(&self.runtime, instance_key)? {
558            return Ok(());
559        }
560        schedule_module_supervision(&self.runtime, instance_key).map_err(|error| {
561            handle_supervision_schedule_failure(&self.runtime, instance_key, error)
562        })
563    }
564
565    /// Starts shutdown admission closure and cooperative cancellation.
566    pub fn request_shutdown(&self) {
567        self.runtime.begin_shutdown();
568    }
569
570    /// Performs bounded graceful shutdown using one global deadline.
571    pub async fn shutdown(&self, timeout: Duration) -> ShutdownOutcome {
572        self.runtime.begin_shutdown();
573        let cleanup_started_at = (self.runtime.driver.now)();
574        if self.runtime.shutdown.start(cleanup_started_at) {
575            self.runtime
576                .diagnostics
577                .emit(DiagnosticSource::Shutdown, cleanup_started_at, |_| {
578                    DiagnosticEvent::ShutdownCleanupStarted { timeout }
579                });
580            let runtime = self.runtime.clone();
581            let worker_runtime = runtime.clone();
582            match (runtime.driver.spawn_local)(Box::pin(async move {
583                let outcome = shutdown_native_modules(&worker_runtime, timeout).await;
584                worker_runtime.complete_shutdown(outcome);
585            })) {
586                Ok(task) => {
587                    runtime.shutdown_task.replace(Some(task));
588                }
589                Err(error) => runtime.complete_shutdown(ShutdownOutcome::RuntimeFailure {
590                    error: RuntimeFailure::Internal {
591                        detail: format!("failed to schedule App shutdown: {error:?}"),
592                    },
593                }),
594            }
595        }
596        self.runtime.shutdown.wait().await
597    }
598
599    /// Invokes a generated request Operation through the caller's resolved binding.
600    pub async fn invoke<C: RequestCapability>(
601        &self,
602        caller_instance: &str,
603        operation: &str,
604        request: C::Request,
605    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
606        self.handle::<C>(caller_instance)?
607            .invoke(operation, request)
608            .await
609    }
610
611    /// Creates a request context with a fresh Kernel Request ID.
612    ///
613    /// `deadline` is an absolute instant returned by the selected
614    /// [`RuntimeDriver`]'s monotonic clock.
615    pub fn invocation_context(
616        &self,
617        deadline: Option<Duration>,
618        cancellation: CancellationToken,
619    ) -> InvocationContext {
620        InvocationContext::new(self.next_request_id(), deadline, cancellation)
621    }
622
623    /// Creates a request context whose deadline is relative to the Driver's clock.
624    pub fn invocation_context_after(
625        &self,
626        timeout: Duration,
627        cancellation: CancellationToken,
628    ) -> InvocationContext {
629        self.invocation_context(
630            Some((self.runtime.driver.now)().saturating_add(timeout)),
631            cancellation,
632        )
633    }
634
635    /// Invokes a request with an explicit propagated Invocation Context.
636    pub async fn invoke_with_context<C: RequestCapability>(
637        &self,
638        caller_instance: &str,
639        operation: &str,
640        context: InvocationContext,
641        request: C::Request,
642    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
643        self.handle::<C>(caller_instance)?
644            .invoke_with_context(operation, context, request)
645            .await
646    }
647
648    /// Materializes one typed bidirectional stream handle from the resolved Plan.
649    pub fn stream_handle<C: StreamCapability>(
650        &self,
651        caller_instance: &str,
652    ) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
653        if self.runtime.admission.is_closed() {
654            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
655        }
656        let Some(endpoints) = self
657            .stream_endpoints::<C>(caller_instance)
658            .filter(|endpoints| !endpoints.is_empty())
659        else {
660            return self.diagnostic_failure(
661                Some(caller_instance),
662                RuntimeFailure::Unavailable { capability: C::ID },
663            );
664        };
665        Ok(NativeStreamHandle::from_endpoints(
666            endpoints,
667            self.runtime.clone(),
668            caller_instance,
669            false,
670        ))
671    }
672
673    /// Materializes an optional typed bidirectional stream handle.
674    pub fn optional_stream_handle<C: StreamCapability>(
675        &self,
676        caller_instance: &str,
677    ) -> Option<NativeStreamHandle<C>> {
678        let caller_instance = caller_instance.to_owned();
679        self.stream_endpoints::<C>(&caller_instance)
680            .filter(|endpoints| !endpoints.is_empty())
681            .map(|endpoints| {
682                NativeStreamHandle::from_endpoints(
683                    endpoints,
684                    self.runtime.clone(),
685                    &caller_instance,
686                    false,
687                )
688            })
689    }
690
691    /// Returns the number of immutable stream endpoints bound to one requirement.
692    pub fn stream_binding_count<C: StreamCapability>(&self, caller_instance: &str) -> usize {
693        self.stream_endpoints::<C>(caller_instance)
694            .map_or(0, <[_]>::len)
695    }
696
697    /// Materializes a typed Event handle and requires at least one subscriber.
698    pub fn event_handle<C: EventCapability>(
699        &self,
700        caller_instance: &str,
701    ) -> Result<NativeEventHandle<C>, RuntimeFailure> {
702        if self.runtime.admission.is_closed() {
703            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
704        }
705        let Some(endpoints) = self
706            .event_endpoints::<C>(caller_instance)
707            .filter(|endpoints| !endpoints.is_empty())
708        else {
709            return self.diagnostic_failure(
710                Some(caller_instance),
711                RuntimeFailure::Unavailable { capability: C::ID },
712            );
713        };
714        Ok(NativeEventHandle::from_endpoints(
715            endpoints,
716            self.runtime.clone(),
717            caller_instance,
718            false,
719        ))
720    }
721
722    /// Materializes an optional typed Event handle.
723    pub fn optional_event_handle<C: EventCapability>(
724        &self,
725        caller_instance: &str,
726    ) -> Option<NativeEventHandle<C>> {
727        let caller_instance = caller_instance.to_owned();
728        self.event_endpoints::<C>(&caller_instance)
729            .filter(|endpoints| !endpoints.is_empty())
730            .map(|endpoints| {
731                NativeEventHandle::from_endpoints(
732                    endpoints,
733                    self.runtime.clone(),
734                    &caller_instance,
735                    false,
736                )
737            })
738    }
739
740    /// Materializes a typed Event handle whose endpoint set may be empty.
741    pub fn many_event_handle<C: EventCapability>(
742        &self,
743        caller_instance: &str,
744    ) -> Result<NativeEventHandle<C>, RuntimeFailure> {
745        if self.runtime.admission.is_closed() {
746            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
747        }
748        let endpoints = self.event_endpoints::<C>(caller_instance).unwrap_or(&[]);
749        Ok(NativeEventHandle::from_endpoints(
750            endpoints,
751            self.runtime.clone(),
752            caller_instance,
753            false,
754        ))
755    }
756
757    /// Returns the number of immutable Event subscriber endpoints bound to a requirement.
758    pub fn event_binding_count<C: EventCapability>(&self, caller_instance: &str) -> usize {
759        self.event_endpoints::<C>(caller_instance)
760            .map_or(0, <[_]>::len)
761    }
762
763    pub(super) fn next_request_id(&self) -> RequestId {
764        let request_id = self.runtime.request_ids.get();
765        self.runtime.request_ids.set(request_id.saturating_add(1));
766        request_id
767    }
768
769    pub(super) fn endpoints<C: RequestCapability>(
770        &self,
771        caller_instance: &str,
772    ) -> Option<&[NativeEndpointBinding]> {
773        self.bindings
774            .get(&(caller_instance.to_owned(), C::ID))
775            .map(Vec::as_slice)
776    }
777
778    pub(super) fn stream_endpoints<C: StreamCapability>(
779        &self,
780        caller_instance: &str,
781    ) -> Option<&[NativeStreamEndpointBinding]> {
782        self.stream_bindings
783            .get(&(caller_instance.to_owned(), C::ID))
784            .map(Vec::as_slice)
785    }
786
787    pub(super) fn event_endpoints<C: EventCapability>(
788        &self,
789        caller_instance: &str,
790    ) -> Option<&[event::NativeEventEndpointBinding]> {
791        self.event_bindings
792            .get(&(caller_instance.to_owned(), C::ID))
793            .map(Vec::as_slice)
794    }
795}
lenso_kernel/runtime.rs

lenso_kernel/
runtime.rs
