1use std::{any::Any, time::Duration};
2
3use super::{
4 CancellationToken, EventCapability, InvocationContext, LocalBoxFuture, NativeAppRuntime,
5 NativeEndpointBinding, NativeEventHandle, NativeRequestEndpoint, NativeRequestHandle,
6 NativeStreamEndpointBinding, NativeStreamHandle, PluginEventDependencyHandle, Rc, RefCell,
7 StreamCapability, Weak,
8};
9
10pub trait RequestCapability: 'static {
11 type Request: 'static;
13 type Response: 'static;
15 type DomainError: 'static;
17 const ID: &'static str;
19 const DESCRIPTOR_VERSION: &'static str;
21
22 #[doc(hidden)]
27 fn invoke_native(
28 endpoint: &dyn NativeRequestEndpoint,
29 operation: &str,
30 request: Self::Request,
31 context: InvocationContext,
32 ) -> NativeRequestFuture<Self>
33 where
34 Self: Sized,
35 {
36 invoke_typed_or_erased_native_request::<Self>(endpoint, operation, request, context)
37 }
38}
39
40#[doc(hidden)]
42pub type NativeRequestFuture<C> = LocalBoxFuture<
43 'static,
44 Result<
45 Result<<C as RequestCapability>::Response, <C as RequestCapability>::DomainError>,
46 RuntimeFailure,
47 >,
48>;
49
50type TypedNativeRequestFn<C> =
51 dyn Fn(&str, <C as RequestCapability>::Request, InvocationContext) -> NativeRequestFuture<C>;
52
53#[doc(hidden)]
59pub struct TypedNativeRequestEndpoint<C: RequestCapability> {
60 invoke: Rc<TypedNativeRequestFn<C>>,
61}
62
63impl<C: RequestCapability> TypedNativeRequestEndpoint<C> {
64 pub fn new(
66 invoke: impl Fn(&str, C::Request, InvocationContext) -> NativeRequestFuture<C> + 'static,
67 ) -> Self {
68 Self {
69 invoke: Rc::new(invoke),
70 }
71 }
72
73 pub fn invoke(
75 &self,
76 operation: &str,
77 request: C::Request,
78 context: InvocationContext,
79 ) -> NativeRequestFuture<C> {
80 (self.invoke)(operation, request, context)
81 }
82}
83
84impl<C: RequestCapability> std::fmt::Debug for TypedNativeRequestEndpoint<C> {
85 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86 formatter
87 .debug_struct("TypedNativeRequestEndpoint")
88 .field("capability", &C::ID)
89 .finish_non_exhaustive()
90 }
91}
92
93#[doc(hidden)]
95pub fn invoke_typed_or_erased_native_request<C: RequestCapability>(
96 endpoint: &dyn NativeRequestEndpoint,
97 operation: &str,
98 request: C::Request,
99 context: InvocationContext,
100) -> NativeRequestFuture<C> {
101 if let Some(endpoint) = endpoint
102 .typed_endpoint()
103 .and_then(|endpoint| endpoint.downcast_ref::<TypedNativeRequestEndpoint<C>>())
104 {
105 endpoint.invoke(operation, request, context)
106 } else {
107 invoke_erased_native_request::<C>(endpoint, operation, request, context)
108 }
109}
110
111#[doc(hidden)]
113pub fn invoke_erased_native_request<C: RequestCapability>(
114 endpoint: &dyn NativeRequestEndpoint,
115 operation: &str,
116 request: C::Request,
117 context: InvocationContext,
118) -> NativeRequestFuture<C> {
119 let invocation = endpoint.invoke(operation, Box::new(request), context);
120 Box::pin(async move {
121 match invocation.await? {
122 Ok(value) => value
123 .downcast::<C::Response>()
124 .map(|value| Ok(*value))
125 .map_err(|_| RuntimeFailure::ProtocolViolation { capability: C::ID }),
126 Err(value) => value
127 .downcast::<C::DomainError>()
128 .map(|value| Err(*value))
129 .map_err(|_| RuntimeFailure::ProtocolViolation { capability: C::ID }),
130 }
131 })
132}
133
134pub type RequestId = u64;
136
137#[derive(Clone, Debug, Eq, PartialEq)]
139pub enum RuntimeFailure {
140 Unavailable { capability: &'static str },
142 UnknownOperation {
144 capability: &'static str,
145 operation: String,
146 },
147 AmbiguousBinding {
149 capability: &'static str,
150 providers: usize,
151 },
152 ProtocolViolation { capability: &'static str },
154 MissingPluginFactory {
156 instance: String,
157 package_id: String,
158 },
159 UnavailableExecutionClass {
161 instance_key: String,
162 execution_class: String,
163 },
164 InvalidResolvedPlan { detail: String },
166 AdmissionClosed,
168 ResourceExhausted {
170 capability: &'static str,
171 operation: String,
172 },
173 DeadlineExceeded { request_id: RequestId },
175 Cancelled { request_id: RequestId },
177 Internal { detail: String },
179 PluginFailure { detail: String },
181 PluginRestartExhausted { instance: String, attempts: usize },
183}
184
185#[derive(Clone, Copy, Debug, Eq, PartialEq)]
187pub enum PluginLifecyclePhase {
188 Prepare,
190 Construct,
192 Activate,
194 Ready,
196 Deactivate,
198}
199
200#[cfg(test)]
201mod typed_endpoint_tests {
202 use std::any::Any;
203
204 use super::*;
205
206 #[derive(Debug)]
207 struct Echo;
208
209 impl RequestCapability for Echo {
210 type Request = u64;
211 type Response = u64;
212 type DomainError = ();
213 const ID: &'static str = "test.echo@1";
214 const DESCRIPTOR_VERSION: &'static str = "1.0.0";
215 }
216
217 #[derive(Debug)]
218 struct Endpoint {
219 typed: TypedNativeRequestEndpoint<Echo>,
220 }
221
222 impl NativeRequestEndpoint for Endpoint {
223 fn capability_id(&self) -> &'static str {
224 Echo::ID
225 }
226
227 fn descriptor_version(&self) -> &'static str {
228 Echo::DESCRIPTOR_VERSION
229 }
230
231 fn operations(&self) -> &'static [&'static str] {
232 &["echo"]
233 }
234
235 fn typed_endpoint(&self) -> Option<&dyn Any> {
236 Some(&self.typed)
237 }
238
239 fn invoke(
240 &self,
241 _operation: &str,
242 _request: Box<dyn Any>,
243 _context: InvocationContext,
244 ) -> LocalBoxFuture<'static, Result<crate::ErasedDomainResult, RuntimeFailure>> {
245 panic!("typed dispatch must not call the erased endpoint")
246 }
247 }
248
249 #[test]
250 fn default_dispatch_uses_runtime_typed_endpoint() {
251 let endpoint = Endpoint {
252 typed: TypedNativeRequestEndpoint::new(|_, request, _| {
253 Box::pin(futures::future::ready(Ok(Ok(request + 1))))
254 }),
255 };
256 let context = InvocationContext::new(1, None, CancellationToken::new());
257
258 let result =
259 futures::executor::block_on(Echo::invoke_native(&endpoint, "echo", 41, context));
260
261 assert_eq!(result, Ok(Ok(42)));
262 }
263}
264
265#[derive(Clone, Debug)]
267pub struct PluginDependency {
268 pub(super) requirement_id: String,
269 pub(super) capability_id: String,
270 pub(super) provider_instance: String,
271 pub(super) provider_order: usize,
272 pub(super) handle: Option<PluginDependencyHandle>,
273 pub(super) stream_handle: Option<PluginStreamDependencyHandle>,
274 pub(super) event_handle: Option<PluginEventDependencyHandle>,
275}
276
277impl PluginDependency {
278 pub(super) fn new(
279 requirement_id: impl Into<String>,
280 capability_id: impl Into<String>,
281 provider_instance: impl Into<String>,
282 provider_order: usize,
283 handle: Option<PluginDependencyHandle>,
284 stream_handle: Option<PluginStreamDependencyHandle>,
285 event_handle: Option<PluginEventDependencyHandle>,
286 ) -> Self {
287 Self {
288 requirement_id: requirement_id.into(),
289 capability_id: capability_id.into(),
290 provider_instance: provider_instance.into(),
291 provider_order,
292 handle,
293 stream_handle,
294 event_handle,
295 }
296 }
297
298 pub fn requirement_id(&self) -> &str {
300 &self.requirement_id
301 }
302
303 pub fn capability_id(&self) -> &str {
305 &self.capability_id
306 }
307
308 pub fn provider_instance(&self) -> &str {
310 &self.provider_instance
311 }
312
313 pub const fn provider_order(&self) -> usize {
315 self.provider_order
316 }
317
318 pub fn handle(&self) -> Option<PluginDependencyHandle> {
320 self.handle.clone()
321 }
322
323 pub fn stream_handle(&self) -> Option<PluginStreamDependencyHandle> {
325 self.stream_handle.clone()
326 }
327
328 pub fn event_handle(&self) -> Option<PluginEventDependencyHandle> {
330 self.event_handle.clone()
331 }
332}
333
334#[derive(Clone, Debug)]
336pub struct PluginDependencyHandle {
337 pub(super) binding: NativeEndpointBinding,
338 pub(super) caller_instance: String,
339 pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
340}
341
342#[derive(Clone, Debug)]
344pub struct PluginStreamDependencyHandle {
345 pub(super) binding: NativeStreamEndpointBinding,
346 pub(super) caller_instance: String,
347 pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
348}
349
350impl PluginStreamDependencyHandle {
351 pub fn capability_id(&self) -> &'static str {
353 self.binding.state.capability_id
354 }
355
356 pub fn descriptor_version(&self) -> &'static str {
358 self.binding.state.descriptor_version
359 }
360
361 pub fn operations(&self) -> &'static [&'static str] {
363 self.binding.state.operations
364 }
365
366 pub fn typed<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
368 if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
369 return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
370 }
371 let runtime = self
372 .runtime
373 .borrow()
374 .upgrade()
375 .ok_or(RuntimeFailure::AdmissionClosed)?;
376 Ok(NativeStreamHandle::from_endpoints(
377 std::slice::from_ref(&self.binding),
378 runtime,
379 &self.caller_instance,
380 true,
381 ))
382 }
383}
384
385impl PluginDependencyHandle {
386 pub fn capability_id(&self) -> &'static str {
388 self.binding.state.capability_id
389 }
390
391 pub fn descriptor_version(&self) -> &'static str {
393 self.binding.state.descriptor_version
394 }
395
396 pub fn operations(&self) -> &'static [&'static str] {
398 self.binding.state.operations
399 }
400
401 pub fn invoke_erased(
406 &self,
407 operation: &str,
408 request: Box<dyn Any>,
409 context: InvocationContext,
410 ) -> LocalBoxFuture<'static, Result<crate::ErasedDomainResult, RuntimeFailure>> {
411 let Some(runtime) = self.runtime.borrow().upgrade() else {
412 return Box::pin(futures::future::ready(Err(RuntimeFailure::AdmissionClosed)));
413 };
414 crate::request_handle::invoke_erased_dependency(
415 self.binding.clone(),
416 runtime,
417 self.caller_instance.clone(),
418 operation.to_owned(),
419 context,
420 request,
421 )
422 }
423
424 pub fn typed<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
426 if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
427 return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
428 }
429 let runtime = self
430 .runtime
431 .borrow()
432 .upgrade()
433 .ok_or(RuntimeFailure::AdmissionClosed)?;
434 Ok(NativeRequestHandle::from_endpoints(
435 std::slice::from_ref(&self.binding),
436 runtime,
437 &self.caller_instance,
438 true,
439 ))
440 }
441}
442
443#[derive(Clone, Debug, Default)]
445pub struct PluginDependencies {
446 pub(super) requirements: Vec<lenso_app_plan::CapabilityRequirementPlan>,
447 pub(super) bindings: Vec<PluginDependency>,
448 pub(super) caller_instance: Rc<str>,
449 pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
450}
451
452impl PluginDependencies {
453 pub(super) fn new(
454 caller_instance: impl Into<String>,
455 runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
456 requirements: Vec<lenso_app_plan::CapabilityRequirementPlan>,
457 ) -> Self {
458 Self {
459 requirements,
460 bindings: Vec::new(),
461 caller_instance: Rc::from(caller_instance.into()),
462 runtime,
463 }
464 }
465
466 pub fn requirements(&self) -> &[lenso_app_plan::CapabilityRequirementPlan] {
468 &self.requirements
469 }
470
471 pub fn requirement(&self, id: &str) -> Result<Self, RuntimeFailure> {
473 let requirement = self
474 .requirements
475 .iter()
476 .find(|requirement| requirement.requirement_id() == id)
477 .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
478 detail: format!(
479 "consumer `{}` has no requirement `{id}`",
480 self.caller_instance
481 ),
482 })?;
483 Ok(Self {
484 requirements: vec![requirement.clone()],
485 bindings: self
486 .bindings
487 .iter()
488 .filter(|binding| binding.requirement_id() == id)
489 .cloned()
490 .collect(),
491 caller_instance: self.caller_instance.clone(),
492 runtime: self.runtime.clone(),
493 })
494 }
495
496 fn validate_lookup(
497 &self,
498 capability: &'static str,
499 version: &str,
500 ) -> Result<(), RuntimeFailure> {
501 let declarations = self
502 .requirements
503 .iter()
504 .filter(|requirement| requirement.capability_id() == capability)
505 .collect::<Vec<_>>();
506 match declarations.as_slice() {
507 [] => Err(RuntimeFailure::Unavailable { capability }),
508 [declaration] if declaration.descriptor_version() == version => Ok(()),
509 [_] => Err(RuntimeFailure::ProtocolViolation { capability }),
510 declarations => Err(RuntimeFailure::AmbiguousBinding {
511 capability,
512 providers: declarations.len(),
513 }),
514 }
515 }
516
517 pub fn bindings(&self) -> &[PluginDependency] {
519 &self.bindings
520 }
521
522 pub fn len(&self) -> usize {
524 self.bindings.len()
525 }
526
527 pub fn is_empty(&self) -> bool {
529 self.bindings.is_empty()
530 }
531
532 pub fn invocation_context(
538 &self,
539 deadline: Option<Duration>,
540 cancellation: CancellationToken,
541 ) -> Result<InvocationContext, RuntimeFailure> {
542 let runtime = self
543 .runtime
544 .borrow()
545 .upgrade()
546 .ok_or(RuntimeFailure::AdmissionClosed)?;
547 let request_id = runtime.request_ids.get();
548 runtime.request_ids.set(request_id.saturating_add(1));
549 Ok(InvocationContext::new(request_id, deadline, cancellation)
550 .with_shared_caller_instance(self.caller_instance.clone()))
551 }
552
553 pub fn invocation_context_after(
555 &self,
556 timeout: Duration,
557 cancellation: CancellationToken,
558 ) -> Result<InvocationContext, RuntimeFailure> {
559 let runtime = self
560 .runtime
561 .borrow()
562 .upgrade()
563 .ok_or(RuntimeFailure::AdmissionClosed)?;
564 let deadline = (runtime.driver.now)().saturating_add(timeout);
565 drop(runtime);
566 self.invocation_context(Some(deadline), cancellation)
567 }
568
569 pub fn one<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
571 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
572 let handles: Vec<_> = self
573 .bindings
574 .iter()
575 .filter(|binding| binding.capability_id() == C::ID)
576 .filter_map(PluginDependency::handle)
577 .collect();
578 match handles.as_slice() {
579 [handle] => handle.typed::<C>(),
580 [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
581 handles => Err(RuntimeFailure::AmbiguousBinding {
582 capability: C::ID,
583 providers: handles.len(),
584 }),
585 }
586 }
587
588 pub fn optional<C: RequestCapability>(
590 &self,
591 ) -> Result<Option<NativeRequestHandle<C>>, RuntimeFailure> {
592 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
593 match self
594 .bindings
595 .iter()
596 .filter(|binding| binding.capability_id() == C::ID)
597 .filter_map(PluginDependency::handle)
598 .collect::<Vec<_>>()
599 .as_slice()
600 {
601 [] => Ok(None),
602 [handle] => handle.typed::<C>().map(Some),
603 handles => Err(RuntimeFailure::AmbiguousBinding {
604 capability: C::ID,
605 providers: handles.len(),
606 }),
607 }
608 }
609
610 pub fn many<C: RequestCapability>(
612 &self,
613 ) -> Result<Vec<NativeRequestHandle<C>>, RuntimeFailure> {
614 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
615 self.bindings
616 .iter()
617 .filter(|binding| binding.capability_id() == C::ID)
618 .filter_map(PluginDependency::handle)
619 .map(|handle| handle.typed::<C>())
620 .collect()
621 }
622
623 pub fn one_stream<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
625 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
626 let handles: Vec<_> = self
627 .bindings
628 .iter()
629 .filter(|binding| binding.capability_id() == C::ID)
630 .filter_map(PluginDependency::stream_handle)
631 .collect();
632 match handles.as_slice() {
633 [handle] => handle.typed::<C>(),
634 [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
635 handles => Err(RuntimeFailure::AmbiguousBinding {
636 capability: C::ID,
637 providers: handles.len(),
638 }),
639 }
640 }
641
642 pub fn optional_stream<C: StreamCapability>(
644 &self,
645 ) -> Result<Option<NativeStreamHandle<C>>, RuntimeFailure> {
646 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
647 match self
648 .bindings
649 .iter()
650 .filter(|binding| binding.capability_id() == C::ID)
651 .filter_map(PluginDependency::stream_handle)
652 .collect::<Vec<_>>()
653 .as_slice()
654 {
655 [] => Ok(None),
656 [handle] => handle.typed::<C>().map(Some),
657 handles => Err(RuntimeFailure::AmbiguousBinding {
658 capability: C::ID,
659 providers: handles.len(),
660 }),
661 }
662 }
663
664 pub fn many_stream<C: StreamCapability>(
666 &self,
667 ) -> Result<Vec<NativeStreamHandle<C>>, RuntimeFailure> {
668 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
669 self.bindings
670 .iter()
671 .filter(|binding| binding.capability_id() == C::ID)
672 .filter_map(PluginDependency::stream_handle)
673 .map(|handle| handle.typed::<C>())
674 .collect()
675 }
676
677 pub fn many_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
679 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
680 let handles: Vec<_> = self
681 .bindings
682 .iter()
683 .filter(|binding| binding.capability_id() == C::ID)
684 .filter_map(PluginDependency::event_handle)
685 .collect();
686 if handles.iter().any(|handle| {
687 handle.capability_id() != C::ID || handle.descriptor_version() != C::DESCRIPTOR_VERSION
688 }) {
689 return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
690 }
691 let runtime = self
692 .runtime
693 .borrow()
694 .upgrade()
695 .ok_or(RuntimeFailure::AdmissionClosed)?;
696 let endpoints = handles
697 .iter()
698 .map(|handle| handle.binding.clone())
699 .collect::<Vec<_>>();
700 Ok(NativeEventHandle::from_endpoints(
701 &endpoints,
702 runtime,
703 &self.caller_instance,
704 true,
705 ))
706 }
707
708 pub fn one_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
710 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
711 match self
712 .bindings
713 .iter()
714 .filter(|binding| binding.capability_id() == C::ID)
715 .filter_map(PluginDependency::event_handle)
716 .collect::<Vec<_>>()
717 .as_slice()
718 {
719 [handle] => handle.typed::<C>(),
720 [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
721 handles => Err(RuntimeFailure::AmbiguousBinding {
722 capability: C::ID,
723 providers: handles.len(),
724 }),
725 }
726 }
727
728 pub fn optional_event<C: EventCapability>(
730 &self,
731 ) -> Result<Option<NativeEventHandle<C>>, RuntimeFailure> {
732 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
733 match self
734 .bindings
735 .iter()
736 .filter(|binding| binding.capability_id() == C::ID)
737 .filter_map(PluginDependency::event_handle)
738 .collect::<Vec<_>>()
739 .as_slice()
740 {
741 [] => Ok(None),
742 [handle] => handle.typed::<C>().map(Some),
743 handles => Err(RuntimeFailure::AmbiguousBinding {
744 capability: C::ID,
745 providers: handles.len(),
746 }),
747 }
748 }
749}