1use std::any::Any;
2
3use super::{
4 BTreeMap, BTreeSet, ErasedDomainResult, ErasedValue, ExecutionClassId, InvocationContext,
5 LocalBoxFuture, NativeEventEndpoint, NativeStreamEndpoint, PluginLifecycle, Rc,
6 ResolvedAppPlan, RuntimeFailure,
7};
8
9pub trait NativeRequestEndpoint: std::fmt::Debug {
11 fn capability_id(&self) -> &'static str;
13 fn descriptor_version(&self) -> &'static str;
15 fn operations(&self) -> &'static [&'static str];
17 #[doc(hidden)]
21 fn typed_endpoint(&self) -> Option<&dyn Any> {
22 None
23 }
24 fn invoke(
26 &self,
27 operation: &str,
28 request: ErasedValue,
29 context: InvocationContext,
30 ) -> LocalBoxFuture<'static, Result<ErasedDomainResult, RuntimeFailure>>;
31}
32
33#[derive(Clone, Debug, Default)]
35pub struct NativeEndpointSet {
36 request: Vec<Rc<dyn NativeRequestEndpoint>>,
37 stream: Vec<Rc<dyn NativeStreamEndpoint>>,
38 event: Vec<Rc<dyn NativeEventEndpoint>>,
39}
40
41impl NativeEndpointSet {
42 pub fn new(
44 request: Vec<Rc<dyn NativeRequestEndpoint>>,
45 stream: Vec<Rc<dyn NativeStreamEndpoint>>,
46 event: Vec<Rc<dyn NativeEventEndpoint>>,
47 ) -> Self {
48 Self {
49 request,
50 stream,
51 event,
52 }
53 }
54
55 pub fn request(&self) -> &[Rc<dyn NativeRequestEndpoint>] {
57 &self.request
58 }
59
60 pub fn stream(&self) -> &[Rc<dyn NativeStreamEndpoint>] {
62 &self.stream
63 }
64
65 pub fn event(&self) -> &[Rc<dyn NativeEventEndpoint>] {
67 &self.event
68 }
69}
70
71#[derive(Debug)]
73pub struct PreparedNativePlugin {
74 pub(super) endpoints: NativeEndpointSet,
75 pub(super) lifecycle: Rc<dyn PluginLifecycle>,
76}
77
78impl PreparedNativePlugin {
79 pub fn new(
81 endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
82 lifecycle: impl PluginLifecycle,
83 ) -> Self {
84 Self {
85 endpoints: NativeEndpointSet::new(endpoints, Vec::new(), Vec::new()),
86 lifecycle: Rc::new(lifecycle),
87 }
88 }
89
90 pub fn with_lifecycle(
92 endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
93 lifecycle: Rc<dyn PluginLifecycle>,
94 ) -> Self {
95 Self {
96 endpoints: NativeEndpointSet::new(endpoints, Vec::new(), Vec::new()),
97 lifecycle,
98 }
99 }
100
101 pub fn with_endpoints(
103 endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
104 stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
105 lifecycle: impl PluginLifecycle,
106 ) -> Self {
107 Self {
108 endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, Vec::new()),
109 lifecycle: Rc::new(lifecycle),
110 }
111 }
112
113 pub fn with_endpoint_set_lifecycle(
115 endpoints: NativeEndpointSet,
116 lifecycle: Rc<dyn PluginLifecycle>,
117 ) -> Self {
118 Self {
119 endpoints,
120 lifecycle,
121 }
122 }
123
124 pub fn with_stream_endpoints(
126 stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
127 lifecycle: impl PluginLifecycle,
128 ) -> Self {
129 Self::with_endpoints(Vec::new(), stream_endpoints, lifecycle)
130 }
131
132 pub fn with_event_endpoints(
134 event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
135 lifecycle: impl PluginLifecycle,
136 ) -> Self {
137 Self::with_all_endpoints(Vec::new(), Vec::new(), event_endpoints, lifecycle)
138 }
139
140 pub fn with_all_endpoints(
142 endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
143 stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
144 event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
145 lifecycle: impl PluginLifecycle,
146 ) -> Self {
147 Self {
148 endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, event_endpoints),
149 lifecycle: Rc::new(lifecycle),
150 }
151 }
152
153 pub fn endpoints(&self) -> &[Rc<dyn NativeRequestEndpoint>] {
155 self.endpoints.request()
156 }
157
158 pub fn stream_endpoints(&self) -> &[Rc<dyn NativeStreamEndpoint>] {
160 self.endpoints.stream()
161 }
162
163 pub fn event_endpoints(&self) -> &[Rc<dyn NativeEventEndpoint>] {
165 self.endpoints.event()
166 }
167
168 pub fn lifecycle(&self) -> Rc<dyn PluginLifecycle> {
170 self.lifecycle.clone()
171 }
172
173 pub(super) fn into_parts(self) -> (NativeEndpointSet, Rc<dyn PluginLifecycle>) {
174 (self.endpoints, self.lifecycle)
175 }
176}
177
178#[derive(Clone, Debug)]
180pub struct PreparedBinding {
181 pub(super) requirement_id: String,
182 pub(super) consumer_instance: String,
183 pub(super) provider_instance: String,
184 pub(super) endpoint: Rc<dyn NativeRequestEndpoint>,
185}
186
187#[derive(Clone, Debug)]
189pub struct PreparedStreamBinding {
190 pub(super) requirement_id: String,
191 pub(super) consumer_instance: String,
192 pub(super) provider_instance: String,
193 pub(super) endpoint: Rc<dyn NativeStreamEndpoint>,
194}
195
196#[derive(Clone, Debug)]
198pub struct PreparedEventBinding {
199 pub(super) requirement_id: String,
200 pub(super) consumer_instance: String,
201 pub(super) provider_instance: String,
202 pub(super) endpoint: Rc<dyn NativeEventEndpoint>,
203}
204
205impl PreparedEventBinding {
206 pub fn new(
208 consumer_instance: impl Into<String>,
209 provider_instance: impl Into<String>,
210 endpoint: Rc<dyn NativeEventEndpoint>,
211 ) -> Self {
212 Self {
213 requirement_id: format!("~{}", endpoint.capability_id()),
214 consumer_instance: consumer_instance.into(),
215 provider_instance: provider_instance.into(),
216 endpoint,
217 }
218 }
219
220 #[must_use]
222 pub fn with_requirement_id(mut self, id: impl Into<String>) -> Self {
223 self.requirement_id = id.into();
224 self
225 }
226
227 pub fn requirement_id(&self) -> &str {
229 &self.requirement_id
230 }
231
232 pub fn consumer_instance(&self) -> &str {
234 &self.consumer_instance
235 }
236
237 pub fn provider_instance(&self) -> &str {
239 &self.provider_instance
240 }
241
242 pub fn endpoint(&self) -> Rc<dyn NativeEventEndpoint> {
244 self.endpoint.clone()
245 }
246
247 pub(super) fn same_identity(&self, other: &Self) -> bool {
248 self.requirement_id == other.requirement_id
249 && self.consumer_instance == other.consumer_instance
250 && self.provider_instance == other.provider_instance
251 && self.endpoint.capability_id() == other.endpoint.capability_id()
252 }
253}
254
255impl PreparedStreamBinding {
256 pub fn new(
258 consumer_instance: impl Into<String>,
259 provider_instance: impl Into<String>,
260 endpoint: Rc<dyn NativeStreamEndpoint>,
261 ) -> Self {
262 Self {
263 requirement_id: format!("~{}", endpoint.capability_id()),
264 consumer_instance: consumer_instance.into(),
265 provider_instance: provider_instance.into(),
266 endpoint,
267 }
268 }
269
270 #[must_use]
272 pub fn with_requirement_id(mut self, id: impl Into<String>) -> Self {
273 self.requirement_id = id.into();
274 self
275 }
276
277 pub fn requirement_id(&self) -> &str {
279 &self.requirement_id
280 }
281
282 pub fn consumer_instance(&self) -> &str {
284 &self.consumer_instance
285 }
286
287 pub fn provider_instance(&self) -> &str {
289 &self.provider_instance
290 }
291
292 pub fn endpoint(&self) -> Rc<dyn NativeStreamEndpoint> {
294 self.endpoint.clone()
295 }
296
297 pub(super) fn same_identity(&self, other: &Self) -> bool {
298 self.requirement_id == other.requirement_id
299 && self.consumer_instance == other.consumer_instance
300 && self.provider_instance == other.provider_instance
301 && self.endpoint.capability_id() == other.endpoint.capability_id()
302 }
303}
304
305impl PreparedBinding {
306 pub fn new(
308 consumer_instance: impl Into<String>,
309 provider_instance: impl Into<String>,
310 endpoint: Rc<dyn NativeRequestEndpoint>,
311 ) -> Self {
312 Self {
313 requirement_id: format!("~{}", endpoint.capability_id()),
314 consumer_instance: consumer_instance.into(),
315 provider_instance: provider_instance.into(),
316 endpoint,
317 }
318 }
319
320 #[must_use]
322 pub fn with_requirement_id(mut self, id: impl Into<String>) -> Self {
323 self.requirement_id = id.into();
324 self
325 }
326
327 pub fn requirement_id(&self) -> &str {
329 &self.requirement_id
330 }
331
332 pub fn consumer_instance(&self) -> &str {
334 &self.consumer_instance
335 }
336
337 pub fn provider_instance(&self) -> &str {
339 &self.provider_instance
340 }
341
342 pub fn endpoint(&self) -> Rc<dyn NativeRequestEndpoint> {
344 self.endpoint.clone()
345 }
346
347 pub(super) fn same_identity(&self, other: &Self) -> bool {
348 self.requirement_id == other.requirement_id
349 && self.consumer_instance == other.consumer_instance
350 && self.provider_instance == other.provider_instance
351 && self.endpoint.capability_id() == other.endpoint.capability_id()
352 }
353}
354
355#[derive(Debug)]
357pub struct PreparedNativeApp {
358 pub(super) bindings: Vec<PreparedBinding>,
359 pub(super) stream_bindings: Vec<PreparedStreamBinding>,
360 pub(super) event_bindings: Vec<PreparedEventBinding>,
361 pub(super) generations: BTreeMap<String, PreparedNativePlugin>,
362}
363
364impl PreparedNativeApp {
365 pub fn new(
367 bindings: Vec<PreparedBinding>,
368 generations: BTreeMap<String, PreparedNativePlugin>,
369 ) -> Self {
370 Self {
371 bindings,
372 stream_bindings: Vec::new(),
373 event_bindings: Vec::new(),
374 generations,
375 }
376 }
377
378 pub fn empty() -> Self {
380 Self::new(Vec::new(), BTreeMap::new())
381 }
382
383 #[must_use]
385 pub fn with_stream_bindings(mut self, stream_bindings: Vec<PreparedStreamBinding>) -> Self {
386 self.stream_bindings = stream_bindings;
387 self
388 }
389
390 #[must_use]
392 pub fn with_event_bindings(mut self, event_bindings: Vec<PreparedEventBinding>) -> Self {
393 self.event_bindings = event_bindings;
394 self
395 }
396
397 pub(super) fn merge(&mut self, other: Self) -> Result<(), RuntimeFailure> {
398 for binding in other.bindings {
399 if self
400 .bindings
401 .iter()
402 .any(|existing| existing.same_identity(&binding))
403 {
404 return Err(RuntimeFailure::InvalidResolvedPlan {
405 detail: format!(
406 "multiple Execution Adapters prepared binding `{}:{}:{}`",
407 binding.consumer_instance,
408 binding.endpoint.capability_id(),
409 binding.provider_instance
410 ),
411 });
412 }
413 self.bindings.push(binding);
414 }
415 for binding in other.stream_bindings {
416 if self
417 .stream_bindings
418 .iter()
419 .any(|existing| existing.same_identity(&binding))
420 {
421 return Err(RuntimeFailure::InvalidResolvedPlan {
422 detail: format!(
423 "multiple Execution Adapters prepared stream binding `{}:{}:{}`",
424 binding.consumer_instance,
425 binding.endpoint.capability_id(),
426 binding.provider_instance
427 ),
428 });
429 }
430 self.stream_bindings.push(binding);
431 }
432 for binding in other.event_bindings {
433 if self
434 .event_bindings
435 .iter()
436 .any(|existing| existing.same_identity(&binding))
437 {
438 return Err(RuntimeFailure::InvalidResolvedPlan {
439 detail: format!(
440 "multiple Execution Adapters prepared Event binding `{}:{}:{}`",
441 binding.consumer_instance,
442 binding.endpoint.capability_id(),
443 binding.provider_instance
444 ),
445 });
446 }
447 self.event_bindings.push(binding);
448 }
449 for (instance_key, generation) in other.generations {
450 if self
451 .generations
452 .insert(instance_key.clone(), generation)
453 .is_some()
454 {
455 return Err(RuntimeFailure::InvalidResolvedPlan {
456 detail: format!(
457 "multiple Execution Adapters prepared Plugin Instance generation `{instance_key}`"
458 ),
459 });
460 }
461 }
462 Ok(())
463 }
464}
465
466pub trait ExecutionAdapter: std::fmt::Debug + 'static {
468 fn supports_plan_transition(&self, _instance: &lenso_app_plan::PluginInstancePlan) -> bool {
472 false
473 }
474 fn supports_runtime_profile(&self, authoring_version: u32, profile: &str) -> bool {
476 authoring_version == 1
477 && (profile == self.execution_class().as_str()
478 || (self.execution_class() == ExecutionClassId::native_rust()
479 && profile == "lenso.native-authoring@1"))
480 }
481 fn execution_class(&self) -> ExecutionClassId;
483
484 fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure>;
486
487 fn recreate(
493 &self,
494 _plan: &ResolvedAppPlan,
495 instance_key: &str,
496 ) -> Result<PreparedNativePlugin, RuntimeFailure> {
497 Err(RuntimeFailure::Internal {
498 detail: format!("Execution Adapter cannot recreate Plugin Instance `{instance_key}`"),
499 })
500 }
501}
502
503pub trait NativeExecutionAdapter: std::fmt::Debug + 'static {
508 fn supports_plan_transition(&self, _instance: &lenso_app_plan::PluginInstancePlan) -> bool {
510 false
511 }
512 fn supports_runtime_profile(&self, authoring_version: u32, profile: &str) -> bool {
514 authoring_version == 1
515 && matches!(profile, "lenso.native-authoring@1" | "lenso.native-rust@1")
516 }
517 fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure>;
519
520 fn recreate(
522 &self,
523 _plan: &ResolvedAppPlan,
524 instance_key: &str,
525 ) -> Result<PreparedNativePlugin, RuntimeFailure> {
526 Err(RuntimeFailure::Internal {
527 detail: format!("Execution Adapter cannot recreate Plugin Instance `{instance_key}`"),
528 })
529 }
530}
531
532impl<T: NativeExecutionAdapter> ExecutionAdapter for T {
533 fn supports_plan_transition(&self, instance: &lenso_app_plan::PluginInstancePlan) -> bool {
534 NativeExecutionAdapter::supports_plan_transition(self, instance)
535 }
536 fn supports_runtime_profile(&self, authoring_version: u32, profile: &str) -> bool {
537 NativeExecutionAdapter::supports_runtime_profile(self, authoring_version, profile)
538 }
539 fn execution_class(&self) -> ExecutionClassId {
540 ExecutionClassId::native_rust()
541 }
542
543 fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure> {
544 NativeExecutionAdapter::prepare(self, plan)
545 }
546
547 fn recreate(
548 &self,
549 plan: &ResolvedAppPlan,
550 instance_key: &str,
551 ) -> Result<PreparedNativePlugin, RuntimeFailure> {
552 NativeExecutionAdapter::recreate(self, plan, instance_key)
553 }
554}
555
556#[derive(Clone, Debug, Default, Eq, PartialEq)]
558pub struct ExecutionClassSet(BTreeSet<ExecutionClassId>);
559
560impl ExecutionClassSet {
561 pub fn contains(&self, execution_class: &ExecutionClassId) -> bool {
563 self.0.contains(execution_class)
564 }
565
566 pub fn iter(&self) -> impl Iterator<Item = &ExecutionClassId> {
568 self.0.iter()
569 }
570}
571
572#[derive(Clone, Debug, Eq, PartialEq)]
574pub enum ExecutionAdapterCatalogError {
575 DuplicateExecutionClass { execution_class: String },
577}
578
579impl std::fmt::Display for ExecutionAdapterCatalogError {
580 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
581 match self {
582 Self::DuplicateExecutionClass { execution_class } => write!(
583 formatter,
584 "multiple Execution Adapters provide class `{execution_class}`"
585 ),
586 }
587 }
588}
589
590impl std::error::Error for ExecutionAdapterCatalogError {}
591
592#[derive(Debug, Default)]
594pub struct ExecutionAdapterCatalog {
595 pub(super) adapters: BTreeMap<ExecutionClassId, Rc<dyn ExecutionAdapter>>,
596}
597
598impl ExecutionAdapterCatalog {
599 pub fn with_stateless_transitions(
605 mut self,
606 successor: &ResolvedAppPlan,
607 instance_keys: &[String],
608 ) -> Result<Self, RuntimeFailure> {
609 successor
610 .validate()
611 .map_err(|_| RuntimeFailure::InvalidResolvedPlan {
612 detail: "invalid transition admission snapshot".into(),
613 })?;
614 let mut admitted =
615 BTreeMap::<ExecutionClassId, Vec<lenso_app_plan::PluginInstancePlan>>::new();
616 for key in instance_keys {
617 let instance = successor.plugin_instance(key).ok_or_else(|| {
618 RuntimeFailure::InvalidResolvedPlan {
619 detail: format!("unknown transition admission Instance `{key}`"),
620 }
621 })?;
622 if !self.adapters.contains_key(instance.execution_class()) {
623 return Err(RuntimeFailure::UnavailableExecutionClass {
624 instance_key: key.clone(),
625 execution_class: instance.execution_class().to_string(),
626 });
627 }
628 admitted
629 .entry(instance.execution_class().clone())
630 .or_default()
631 .push(instance.clone());
632 }
633 for (class, instances) in admitted {
634 let inner = self.adapters.remove(&class).expect("admitted class exists");
635 self.adapters.insert(
636 class,
637 Rc::new(TransitionAdmittedAdapter { inner, instances }),
638 );
639 }
640 Ok(self)
641 }
642 pub fn new() -> Self {
644 Self::default()
645 }
646
647 pub fn single(adapter: impl ExecutionAdapter) -> Self {
649 Self::new()
650 .with_adapter(adapter)
651 .expect("a new catalog cannot contain a duplicate execution class")
652 }
653
654 pub fn with_adapter(
656 self,
657 adapter: impl ExecutionAdapter,
658 ) -> Result<Self, ExecutionAdapterCatalogError> {
659 self.with_shared_adapter(Rc::new(adapter))
660 }
661
662 pub fn with_shared_adapter(
664 mut self,
665 adapter: Rc<dyn ExecutionAdapter>,
666 ) -> Result<Self, ExecutionAdapterCatalogError> {
667 let execution_class = adapter.execution_class();
668 if self.adapters.contains_key(&execution_class) {
669 return Err(ExecutionAdapterCatalogError::DuplicateExecutionClass {
670 execution_class: execution_class.to_string(),
671 });
672 }
673 self.adapters.insert(execution_class, adapter);
674 Ok(self)
675 }
676
677 pub fn execution_classes(&self) -> ExecutionClassSet {
679 ExecutionClassSet(self.adapters.keys().cloned().collect())
680 }
681
682 pub(super) fn adapter(
683 &self,
684 execution_class: &ExecutionClassId,
685 ) -> Option<Rc<dyn ExecutionAdapter>> {
686 self.adapters.get(execution_class).cloned()
687 }
688
689 pub(super) fn prepare(
690 &self,
691 plan: &ResolvedAppPlan,
692 ) -> Result<PreparedNativeApp, RuntimeFailure> {
693 let mut required_classes = BTreeSet::new();
694 for instance in plan.plugin_instances() {
695 if !self.adapters.contains_key(instance.execution_class()) {
696 return Err(RuntimeFailure::UnavailableExecutionClass {
697 instance_key: instance.instance_key().to_owned(),
698 execution_class: instance.execution_class().to_string(),
699 });
700 }
701 required_classes.insert(instance.execution_class().clone());
702 if !self.adapters[instance.execution_class()]
703 .supports_runtime_profile(instance.authoring_version(), instance.runtime_profile())
704 {
705 return Err(RuntimeFailure::InvalidResolvedPlan {
706 detail: format!(
707 "Adapter `{}` does not support authoring {} profile `{}` for `{}`",
708 instance.execution_class(),
709 instance.authoring_version(),
710 instance.runtime_profile(),
711 instance.instance_key()
712 ),
713 });
714 }
715 }
716
717 let mut prepared = PreparedNativeApp::empty();
718 for execution_class in required_classes {
719 let adapter = self
720 .adapters
721 .get(&execution_class)
722 .expect("required execution classes were validated");
723 prepared.merge(adapter.prepare(plan)?)?;
724 }
725 Ok(prepared)
726 }
727}
728
729#[derive(Debug)]
730struct TransitionAdmittedAdapter {
731 inner: Rc<dyn ExecutionAdapter>,
732 instances: Vec<lenso_app_plan::PluginInstancePlan>,
733}
734
735impl ExecutionAdapter for TransitionAdmittedAdapter {
736 fn supports_plan_transition(&self, instance: &lenso_app_plan::PluginInstancePlan) -> bool {
737 self.instances.contains(instance)
738 }
739 fn supports_runtime_profile(&self, version: u32, profile: &str) -> bool {
740 self.inner.supports_runtime_profile(version, profile)
741 }
742 fn execution_class(&self) -> ExecutionClassId {
743 self.inner.execution_class()
744 }
745 fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure> {
746 self.inner.prepare(plan)
747 }
748 fn recreate(
749 &self,
750 plan: &ResolvedAppPlan,
751 key: &str,
752 ) -> Result<PreparedNativePlugin, RuntimeFailure> {
753 self.inner.recreate(plan, key)
754 }
755}