1use std::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 typed<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
403 if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
404 return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
405 }
406 let runtime = self
407 .runtime
408 .borrow()
409 .upgrade()
410 .ok_or(RuntimeFailure::AdmissionClosed)?;
411 Ok(NativeRequestHandle::from_endpoints(
412 std::slice::from_ref(&self.binding),
413 runtime,
414 &self.caller_instance,
415 true,
416 ))
417 }
418}
419
420#[derive(Clone, Debug, Default)]
422pub struct PluginDependencies {
423 pub(super) requirements: Vec<lenso_app_plan::CapabilityRequirementPlan>,
424 pub(super) bindings: Vec<PluginDependency>,
425 pub(super) caller_instance: Rc<str>,
426 pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
427}
428
429impl PluginDependencies {
430 pub(super) fn new(
431 caller_instance: impl Into<String>,
432 runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
433 requirements: Vec<lenso_app_plan::CapabilityRequirementPlan>,
434 ) -> Self {
435 Self {
436 requirements,
437 bindings: Vec::new(),
438 caller_instance: Rc::from(caller_instance.into()),
439 runtime,
440 }
441 }
442
443 pub fn requirements(&self) -> &[lenso_app_plan::CapabilityRequirementPlan] {
445 &self.requirements
446 }
447
448 pub fn requirement(&self, id: &str) -> Result<Self, RuntimeFailure> {
450 let requirement = self
451 .requirements
452 .iter()
453 .find(|requirement| requirement.requirement_id() == id)
454 .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
455 detail: format!(
456 "consumer `{}` has no requirement `{id}`",
457 self.caller_instance
458 ),
459 })?;
460 Ok(Self {
461 requirements: vec![requirement.clone()],
462 bindings: self
463 .bindings
464 .iter()
465 .filter(|binding| binding.requirement_id() == id)
466 .cloned()
467 .collect(),
468 caller_instance: self.caller_instance.clone(),
469 runtime: self.runtime.clone(),
470 })
471 }
472
473 fn validate_lookup(
474 &self,
475 capability: &'static str,
476 version: &str,
477 ) -> Result<(), RuntimeFailure> {
478 let declarations = self
479 .requirements
480 .iter()
481 .filter(|requirement| requirement.capability_id() == capability)
482 .collect::<Vec<_>>();
483 match declarations.as_slice() {
484 [] => Err(RuntimeFailure::Unavailable { capability }),
485 [declaration] if declaration.descriptor_version() == version => Ok(()),
486 [_] => Err(RuntimeFailure::ProtocolViolation { capability }),
487 declarations => Err(RuntimeFailure::AmbiguousBinding {
488 capability,
489 providers: declarations.len(),
490 }),
491 }
492 }
493
494 pub fn bindings(&self) -> &[PluginDependency] {
496 &self.bindings
497 }
498
499 pub fn len(&self) -> usize {
501 self.bindings.len()
502 }
503
504 pub fn is_empty(&self) -> bool {
506 self.bindings.is_empty()
507 }
508
509 pub fn invocation_context(
515 &self,
516 deadline: Option<Duration>,
517 cancellation: CancellationToken,
518 ) -> Result<InvocationContext, RuntimeFailure> {
519 let runtime = self
520 .runtime
521 .borrow()
522 .upgrade()
523 .ok_or(RuntimeFailure::AdmissionClosed)?;
524 let request_id = runtime.request_ids.get();
525 runtime.request_ids.set(request_id.saturating_add(1));
526 Ok(InvocationContext::new(request_id, deadline, cancellation)
527 .with_shared_caller_instance(self.caller_instance.clone()))
528 }
529
530 pub fn invocation_context_after(
532 &self,
533 timeout: Duration,
534 cancellation: CancellationToken,
535 ) -> Result<InvocationContext, RuntimeFailure> {
536 let runtime = self
537 .runtime
538 .borrow()
539 .upgrade()
540 .ok_or(RuntimeFailure::AdmissionClosed)?;
541 let deadline = (runtime.driver.now)().saturating_add(timeout);
542 drop(runtime);
543 self.invocation_context(Some(deadline), cancellation)
544 }
545
546 pub fn one<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
548 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
549 let handles: Vec<_> = self
550 .bindings
551 .iter()
552 .filter(|binding| binding.capability_id() == C::ID)
553 .filter_map(PluginDependency::handle)
554 .collect();
555 match handles.as_slice() {
556 [handle] => handle.typed::<C>(),
557 [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
558 handles => Err(RuntimeFailure::AmbiguousBinding {
559 capability: C::ID,
560 providers: handles.len(),
561 }),
562 }
563 }
564
565 pub fn optional<C: RequestCapability>(
567 &self,
568 ) -> Result<Option<NativeRequestHandle<C>>, RuntimeFailure> {
569 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
570 match self
571 .bindings
572 .iter()
573 .filter(|binding| binding.capability_id() == C::ID)
574 .filter_map(PluginDependency::handle)
575 .collect::<Vec<_>>()
576 .as_slice()
577 {
578 [] => Ok(None),
579 [handle] => handle.typed::<C>().map(Some),
580 handles => Err(RuntimeFailure::AmbiguousBinding {
581 capability: C::ID,
582 providers: handles.len(),
583 }),
584 }
585 }
586
587 pub fn many<C: RequestCapability>(
589 &self,
590 ) -> Result<Vec<NativeRequestHandle<C>>, RuntimeFailure> {
591 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
592 self.bindings
593 .iter()
594 .filter(|binding| binding.capability_id() == C::ID)
595 .filter_map(PluginDependency::handle)
596 .map(|handle| handle.typed::<C>())
597 .collect()
598 }
599
600 pub fn one_stream<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
602 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
603 let handles: Vec<_> = self
604 .bindings
605 .iter()
606 .filter(|binding| binding.capability_id() == C::ID)
607 .filter_map(PluginDependency::stream_handle)
608 .collect();
609 match handles.as_slice() {
610 [handle] => handle.typed::<C>(),
611 [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
612 handles => Err(RuntimeFailure::AmbiguousBinding {
613 capability: C::ID,
614 providers: handles.len(),
615 }),
616 }
617 }
618
619 pub fn optional_stream<C: StreamCapability>(
621 &self,
622 ) -> Result<Option<NativeStreamHandle<C>>, RuntimeFailure> {
623 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
624 match self
625 .bindings
626 .iter()
627 .filter(|binding| binding.capability_id() == C::ID)
628 .filter_map(PluginDependency::stream_handle)
629 .collect::<Vec<_>>()
630 .as_slice()
631 {
632 [] => Ok(None),
633 [handle] => handle.typed::<C>().map(Some),
634 handles => Err(RuntimeFailure::AmbiguousBinding {
635 capability: C::ID,
636 providers: handles.len(),
637 }),
638 }
639 }
640
641 pub fn many_stream<C: StreamCapability>(
643 &self,
644 ) -> Result<Vec<NativeStreamHandle<C>>, RuntimeFailure> {
645 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
646 self.bindings
647 .iter()
648 .filter(|binding| binding.capability_id() == C::ID)
649 .filter_map(PluginDependency::stream_handle)
650 .map(|handle| handle.typed::<C>())
651 .collect()
652 }
653
654 pub fn many_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
656 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
657 let handles: Vec<_> = self
658 .bindings
659 .iter()
660 .filter(|binding| binding.capability_id() == C::ID)
661 .filter_map(PluginDependency::event_handle)
662 .collect();
663 if handles.iter().any(|handle| {
664 handle.capability_id() != C::ID || handle.descriptor_version() != C::DESCRIPTOR_VERSION
665 }) {
666 return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
667 }
668 let runtime = self
669 .runtime
670 .borrow()
671 .upgrade()
672 .ok_or(RuntimeFailure::AdmissionClosed)?;
673 let endpoints = handles
674 .iter()
675 .map(|handle| handle.binding.clone())
676 .collect::<Vec<_>>();
677 Ok(NativeEventHandle::from_endpoints(
678 &endpoints,
679 runtime,
680 &self.caller_instance,
681 true,
682 ))
683 }
684
685 pub fn one_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
687 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
688 match self
689 .bindings
690 .iter()
691 .filter(|binding| binding.capability_id() == C::ID)
692 .filter_map(PluginDependency::event_handle)
693 .collect::<Vec<_>>()
694 .as_slice()
695 {
696 [handle] => handle.typed::<C>(),
697 [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
698 handles => Err(RuntimeFailure::AmbiguousBinding {
699 capability: C::ID,
700 providers: handles.len(),
701 }),
702 }
703 }
704
705 pub fn optional_event<C: EventCapability>(
707 &self,
708 ) -> Result<Option<NativeEventHandle<C>>, RuntimeFailure> {
709 self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
710 match self
711 .bindings
712 .iter()
713 .filter(|binding| binding.capability_id() == C::ID)
714 .filter_map(PluginDependency::event_handle)
715 .collect::<Vec<_>>()
716 .as_slice()
717 {
718 [] => Ok(None),
719 [handle] => handle.typed::<C>().map(Some),
720 handles => Err(RuntimeFailure::AmbiguousBinding {
721 capability: C::ID,
722 providers: handles.len(),
723 }),
724 }
725 }
726}