lenso_native_adapter/
lib.rs

1//! Native Rust Execution Adapter for statically linked Plugin packages.
2
3mod authoring;
4mod managed_tasks;
5
6use std::{
7    collections::BTreeMap,
8    rc::Rc,
9    sync::{Mutex, OnceLock},
10};
11
12#[doc(hidden)]
13pub use authoring::{CompleteObjectLifecycle, ConstructionContext, LifecycleContext, PluginObject};
14#[doc(hidden)]
15pub use inventory as __inventory;
16use lenso_app_plan::{
17    ExecutionClassId, ResolvedAppPlan,
18    authoring::{HostCatalog, HostDefaultPlugin, HostPluginRelease, HostSlot, PluginDescriptor},
19};
20use lenso_kernel::{ActivateContext, DeactivateContext, PrepareContext};
21pub use lenso_kernel::{CancellationToken, RuntimeFailure};
22pub use lenso_native_adapter_macros::{PluginConfig, plugin, plugin_impl, provides};
23pub use lenso_runtime_codec::InstanceResources;
24pub use managed_tasks::{ManagedTasks, ManagedTasksError};
25
26/// Optional convention-based lifecycle hooks for a struct-level Plugin.
27///
28/// Add `#[plugin(lifecycle)]`, implement this trait, and override only the
29/// phases that own real work. The generated Adapter lifecycle still connects
30/// declared Capability ports before `activate`.
31#[allow(async_fn_in_trait)]
32pub trait Lifecycle: Clone + 'static {
33    async fn prepare(&self, _context: PrepareContext) -> Result<(), RuntimeFailure> {
34        Ok(())
35    }
36
37    async fn activate(&self, _context: ActivateContext) -> Result<(), RuntimeFailure> {
38        Ok(())
39    }
40
41    async fn deactivate(&self, _context: DeactivateContext) -> Result<(), RuntimeFailure> {
42        Ok(())
43    }
44}
45
46/// Implementation details referenced by generated Plugin glue.
47#[doc(hidden)]
48pub mod __private {
49    pub use crate::authoring::{ErasedConstructionFuture, LinkedPluginConstruction};
50    pub use crate::{
51        __inventory, CompleteObjectLifecycle, ConstructionContext, Lifecycle, LifecycleContext,
52        LinkedNativePluginFactory, NativePluginFactory, NativePluginFactoryContext,
53        NativePluginInstance, PluginObject, RuntimeFailure, link_native_plugin,
54    };
55    pub use futures;
56    pub use futures::future::LocalBoxFuture;
57    pub use lenso_kernel::{
58        ActivateContext, DeactivateContext, InvocationContext, NativeEventEndpoint,
59        NativeRequestEndpoint, NativeRequestFuture, NativeStreamEndpoint, NativeStreamSession,
60        PluginFuture, PluginLifecycle, PrepareContext,
61    };
62    pub use lenso_plugin_authoring::{
63        BoundCapabilityClient, CapabilityClient, CapabilityClientMany,
64    };
65    pub use lenso_runtime_codec::InstanceResources;
66    pub use serde_json;
67}
68
69use lenso_kernel::{
70    NativeEndpointSet, NativeEventEndpoint, NativeExecutionAdapter, NativeRequestEndpoint,
71    NativeStreamEndpoint, NoopPluginLifecycle, PluginLifecycle, PreparedBinding,
72    PreparedEventBinding, PreparedNativeApp, PreparedNativePlugin, PreparedStreamBinding,
73};
74
75/// One native Plugin factory contributed to the Host's link-time catalog.
76#[derive(Clone, Copy, Debug)]
77#[doc(hidden)]
78pub struct LinkedNativePluginFactory {
79    constructor: fn() -> Rc<dyn NativePluginFactory>,
80    descriptor: &'static str,
81}
82
83impl LinkedNativePluginFactory {
84    /// Creates a link-time catalog record. Intended for generated authoring glue.
85    #[doc(hidden)]
86    pub const fn new(
87        constructor: fn() -> Rc<dyn NativePluginFactory>,
88        descriptor: &'static str,
89    ) -> Self {
90        Self {
91            constructor,
92            descriptor,
93        }
94    }
95}
96
97inventory::collect!(LinkedNativePluginFactory);
98
99fn explicitly_linked_factories() -> &'static Mutex<Vec<LinkedNativePluginFactory>> {
100    static FACTORIES: OnceLock<Mutex<Vec<LinkedNativePluginFactory>>> = OnceLock::new();
101    FACTORIES.get_or_init(|| Mutex::new(Vec::new()))
102}
103
104/// Retains one generated native Plugin registration through an explicit Host link call.
105#[doc(hidden)]
106pub fn link_native_plugin(factory: LinkedNativePluginFactory) {
107    let mut factories = explicitly_linked_factories()
108        .lock()
109        .unwrap_or_else(std::sync::PoisonError::into_inner);
110    if !factories.iter().any(|linked| {
111        linked.descriptor == factory.descriptor
112            && std::ptr::fn_addr_eq(linked.constructor, factory.constructor)
113    }) {
114        factories.push(factory);
115    }
116}
117
118fn linked_factories() -> Vec<LinkedNativePluginFactory> {
119    let factories = inventory::iter::<LinkedNativePluginFactory>
120        .into_iter()
121        .copied()
122        .collect::<Vec<_>>();
123    let mut factories = factories;
124    factories.extend(
125        explicitly_linked_factories()
126            .lock()
127            .unwrap_or_else(std::sync::PoisonError::into_inner)
128            .iter()
129            .copied(),
130    );
131    factories
132        .into_iter()
133        .fold(Vec::new(), |mut unique, factory| {
134            if !unique.iter().any(|linked: &LinkedNativePluginFactory| {
135                linked.descriptor == factory.descriptor
136                    && std::ptr::fn_addr_eq(linked.constructor, factory.constructor)
137            }) {
138                unique.push(factory);
139            }
140            unique
141        })
142}
143
144/// Endpoints created for one statically linked Plugin Instance generation.
145#[derive(Debug)]
146pub struct NativePluginInstance {
147    endpoints: NativeEndpointSet,
148    lifecycle: Rc<dyn PluginLifecycle>,
149}
150
151impl NativePluginInstance {
152    /// Creates a generation from its exact declared endpoint set.
153    pub fn new(endpoints: Vec<Rc<dyn NativeRequestEndpoint>>) -> Self {
154        Self::with_lifecycle(endpoints, NoopPluginLifecycle)
155    }
156
157    /// Creates a generation with its exact endpoints and lifecycle Interface.
158    pub fn with_lifecycle(
159        endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
160        lifecycle: impl PluginLifecycle,
161    ) -> Self {
162        Self {
163            endpoints: NativeEndpointSet::new(endpoints, Vec::new(), Vec::new()),
164            lifecycle: Rc::new(lifecycle),
165        }
166    }
167
168    /// Creates a generation with request and bidirectional stream endpoints.
169    pub fn with_endpoints(
170        endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
171        stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
172        lifecycle: impl PluginLifecycle,
173    ) -> Self {
174        Self {
175            endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, Vec::new()),
176            lifecycle: Rc::new(lifecycle),
177        }
178    }
179
180    /// Creates a generation containing only bidirectional stream endpoints.
181    pub fn with_stream_endpoints(
182        stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
183        lifecycle: impl PluginLifecycle,
184    ) -> Self {
185        Self::with_endpoints(Vec::new(), stream_endpoints, lifecycle)
186    }
187
188    /// Creates a generation containing only ephemeral Event endpoints.
189    pub fn with_event_endpoints(
190        event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
191        lifecycle: impl PluginLifecycle,
192    ) -> Self {
193        Self {
194            endpoints: NativeEndpointSet::new(Vec::new(), Vec::new(), event_endpoints),
195            lifecycle: Rc::new(lifecycle),
196        }
197    }
198
199    /// Creates a generation with request, stream, and ephemeral Event endpoints.
200    pub fn with_all_endpoints(
201        endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
202        stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
203        event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
204        lifecycle: impl PluginLifecycle,
205    ) -> Self {
206        Self {
207            endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, event_endpoints),
208            lifecycle: Rc::new(lifecycle),
209        }
210    }
211
212    /// Returns the lifecycle Interface for this generation.
213    pub fn lifecycle(&self) -> Rc<dyn PluginLifecycle> {
214        self.lifecycle.clone()
215    }
216
217    /// Returns the exact endpoint set created for this generation.
218    pub fn endpoints(&self) -> &[Rc<dyn NativeRequestEndpoint>] {
219        self.endpoints.request()
220    }
221
222    /// Returns the exact bidirectional stream endpoint set created for this generation.
223    pub fn stream_endpoints(&self) -> &[Rc<dyn NativeStreamEndpoint>] {
224        self.endpoints.stream()
225    }
226
227    /// Returns the exact ephemeral Event endpoint set created for this Instance.
228    pub fn event_endpoints(&self) -> &[Rc<dyn NativeEventEndpoint>] {
229        self.endpoints.event()
230    }
231}
232
233impl Default for NativePluginInstance {
234    fn default() -> Self {
235        Self::new(Vec::new())
236    }
237}
238
239/// Adapter-specific factory for a statically linked native Rust Plugin.
240pub trait NativePluginFactory: std::fmt::Debug + 'static {
241    /// Package identity selected by the Resolved App Plan.
242    fn package_id(&self) -> &'static str;
243    /// Exact statically linked Cargo package version.
244    fn package_version(&self) -> &'static str {
245        ""
246    }
247    /// Exact authoring/runtime protocol implemented by this factory.
248    fn runtime_profile(&self) -> &'static str {
249        "lenso.native-authoring@1"
250    }
251    /// Immutable factory identity advertised by the exact Host Build Manifest.
252    ///
253    /// Plugin-resolved Plans carry this value as their package revision. The
254    /// default keeps ordinary statically linked factories unique by package and
255    /// version while allowing a factory to override the identity when its build
256    /// authority is more specific than a Cargo package version.
257    fn factory_identity(&self) -> String {
258        let version = self.package_version();
259        if version.is_empty() {
260            self.package_id().to_owned()
261        } else {
262            format!("{}@{version}", self.package_id())
263        }
264    }
265    /// Creates a fresh Plugin Instance generation.
266    fn instantiate(
267        &self,
268        context: NativePluginFactoryContext<'_>,
269    ) -> Result<NativePluginInstance, RuntimeFailure>;
270}
271
272/// Immutable Plan input supplied when a native factory creates one generation.
273#[derive(Clone, Copy, Debug)]
274pub struct NativePluginFactoryContext<'a> {
275    instance_key: &'a str,
276    entrypoint: &'a str,
277    configuration: &'a str,
278    resources: &'a InstanceResources,
279}
280
281impl<'a> NativePluginFactoryContext<'a> {
282    fn from_plan(
283        instance: &'a lenso_app_plan::PluginInstancePlan,
284        resources: &'a InstanceResources,
285    ) -> Self {
286        Self {
287            instance_key: instance.instance_key(),
288            entrypoint: instance.entrypoint(),
289            configuration: instance.configuration(),
290            resources,
291        }
292    }
293
294    /// Returns the App-local Plugin Instance key.
295    pub const fn instance_key(self) -> &'a str {
296        self.instance_key
297    }
298
299    /// Returns the exact package entrypoint selected before boot.
300    pub const fn entrypoint(self) -> &'a str {
301        self.entrypoint
302    }
303
304    /// Returns opaque Plugin-owned configuration selected before boot.
305    pub const fn configuration(self) -> &'a str {
306        self.configuration
307    }
308
309    /// Returns immutable supporting files snapshotted for this Generation.
310    pub const fn resources(self) -> &'a InstanceResources {
311        self.resources
312    }
313}
314
315/// Statically linked native Plugin factories available to an App binary.
316#[derive(Debug, Default)]
317pub struct NativePluginRegistry {
318    factories: Vec<Rc<dyn NativePluginFactory>>,
319    resources: lenso_runtime_codec::InstanceResourceCatalog,
320}
321
322type NativeInstances = BTreeMap<String, NativePluginInstance>;
323type PreparedGenerations = BTreeMap<String, PreparedNativePlugin>;
324type NativeBindings = (
325    Vec<PreparedBinding>,
326    Vec<PreparedStreamBinding>,
327    Vec<PreparedEventBinding>,
328);
329
330fn factory_matches(
331    factory: &dyn NativePluginFactory,
332    expected: &lenso_app_plan::PluginInstancePlan,
333) -> bool {
334    factory.package_id() == expected.package_id()
335        && factory.runtime_profile() == expected.runtime_profile()
336        && (expected.package_revision().is_empty()
337            || factory.package_version() == expected.package_revision()
338            || factory.factory_identity() == expected.package_revision())
339}
340
341impl NativePluginRegistry {
342    /// Creates an empty linked-factory registry.
343    pub fn new() -> Self {
344        Self::default()
345    }
346
347    /// Adds every Plugin factory contributed to this Host at link time.
348    ///
349    /// This catalog describes code available in the binary. The Resolved App
350    /// Plan remains the sole authority that selects and binds Plugin Instances.
351    #[must_use]
352    pub fn with_linked_factories(mut self) -> Self {
353        self.factories.extend(
354            linked_factories()
355                .into_iter()
356                .map(|linked| (linked.constructor)()),
357        );
358        self.factories
359            .sort_by_key(|factory| factory.factory_identity());
360        self.factories
361            .dedup_by(|left, right| left.factory_identity() == right.factory_identity());
362        self
363    }
364
365    /// Injects exact Generation-bound supporting files for selected Instances.
366    #[must_use]
367    pub fn with_resources(
368        mut self,
369        resources: lenso_runtime_codec::InstanceResourceCatalog,
370    ) -> Self {
371        self.resources = resources;
372        self
373    }
374
375    /// Returns the exact native factories available to this registry.
376    pub fn factories(&self) -> impl Iterator<Item = &dyn NativePluginFactory> {
377        self.factories.iter().map(std::convert::AsRef::as_ref)
378    }
379
380    /// Builds the immutable Host Catalog declared by this binary and Host policy.
381    pub fn host_catalog(
382        slots: impl IntoIterator<Item = HostSlot>,
383        defaults: impl IntoIterator<Item = HostDefaultPlugin>,
384    ) -> Result<HostCatalog, RuntimeFailure> {
385        let plugins = linked_factories()
386            .into_iter()
387            .map(|linked| {
388                serde_json::from_str::<PluginDescriptor>(linked.descriptor)
389                    .map(HostPluginRelease::new)
390                    .map_err(|error| RuntimeFailure::InvalidResolvedPlan {
391                        detail: format!("invalid linked Plugin Descriptor: {error}"),
392                    })
393            })
394            .collect::<Result<Vec<_>, _>>()?;
395        Ok(HostCatalog::new(slots, plugins, defaults))
396    }
397    /// Adds one statically linked factory.
398    #[must_use]
399    pub fn with_factory(mut self, factory: impl NativePluginFactory) -> Self {
400        self.factories.push(Rc::new(factory));
401        self
402    }
403
404    fn prepare_instances(
405        &self,
406        plan: &ResolvedAppPlan,
407    ) -> Result<(NativeInstances, PreparedGenerations), RuntimeFailure> {
408        let mut instances = BTreeMap::new();
409        let mut generations = BTreeMap::new();
410        for expected in plan
411            .plugin_instances()
412            .iter()
413            .filter(|instance| instance.execution_class() == &ExecutionClassId::native_rust())
414        {
415            let matching_factories: Vec<_> = self
416                .factories
417                .iter()
418                .filter(|factory| factory_matches(factory.as_ref(), expected))
419                .collect();
420            let factory = match matching_factories.as_slice() {
421                [] => {
422                    return Err(RuntimeFailure::MissingPluginFactory {
423                        instance: expected.instance_key().to_owned(),
424                        package_id: expected.package_id().to_owned(),
425                    });
426                }
427                [factory] => *factory,
428                _ => {
429                    return invalid(format!(
430                        "multiple statically linked factories declare package `{}`",
431                        expected.package_id()
432                    ));
433                }
434            };
435            let generation = factory.instantiate(NativePluginFactoryContext::from_plan(
436                expected,
437                self.resources.for_instance(expected.instance_key()),
438            ))?;
439            generations.insert(
440                expected.instance_key().to_owned(),
441                PreparedNativePlugin::with_endpoint_set_lifecycle(
442                    generation.endpoints.clone(),
443                    generation.lifecycle(),
444                ),
445            );
446            if instances
447                .insert(expected.instance_key().to_owned(), generation)
448                .is_some()
449            {
450                return invalid(format!(
451                    "duplicate Plugin Instance `{}`",
452                    expected.instance_key()
453                ));
454            }
455        }
456        Ok((instances, generations))
457    }
458}
459
460impl NativeExecutionAdapter for NativePluginRegistry {
461    fn supports_runtime_profile(&self, authoring_version: u32, profile: &str) -> bool {
462        matches!(
463            (authoring_version, profile),
464            (1, "lenso.native-authoring@1" | "lenso.native-rust@1")
465                | (2, "lenso.native-authoring@2")
466        )
467    }
468
469    fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure> {
470        plan.validate()
471            .map_err(|error| RuntimeFailure::InvalidResolvedPlan {
472                detail: error.to_string(),
473            })?;
474
475        let (instances, generations) = self.prepare_instances(plan)?;
476        let (bindings, stream_bindings, event_bindings) = prepare_bindings(plan, &instances)?;
477        Ok(PreparedNativeApp::new(bindings, generations)
478            .with_stream_bindings(stream_bindings)
479            .with_event_bindings(event_bindings))
480    }
481
482    fn recreate(
483        &self,
484        plan: &ResolvedAppPlan,
485        instance_key: &str,
486    ) -> Result<PreparedNativePlugin, RuntimeFailure> {
487        let expected = plan
488            .plugin_instances()
489            .iter()
490            .find(|instance| instance.instance_key() == instance_key)
491            .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
492                detail: format!("unknown Plugin Instance `{instance_key}`"),
493            })?;
494        let matching_factories: Vec<_> = self
495            .factories
496            .iter()
497            .filter(|factory| factory_matches(factory.as_ref(), expected))
498            .collect();
499        let factory = match matching_factories.as_slice() {
500            [] => {
501                return Err(RuntimeFailure::MissingPluginFactory {
502                    instance: expected.instance_key().to_owned(),
503                    package_id: expected.package_id().to_owned(),
504                });
505            }
506            [factory] => *factory,
507            _ => {
508                return invalid(format!(
509                    "multiple statically linked factories declare package `{}`",
510                    expected.package_id()
511                ));
512            }
513        };
514        let generation = factory.instantiate(NativePluginFactoryContext::from_plan(
515            expected,
516            self.resources.for_instance(expected.instance_key()),
517        ))?;
518        Ok(PreparedNativePlugin::with_endpoint_set_lifecycle(
519            generation.endpoints.clone(),
520            generation.lifecycle(),
521        ))
522    }
523}
524
525fn prepare_bindings(
526    plan: &ResolvedAppPlan,
527    instances: &NativeInstances,
528) -> Result<NativeBindings, RuntimeFailure> {
529    let mut bindings = Vec::new();
530    let mut stream_bindings = Vec::new();
531    let mut event_bindings = Vec::new();
532    for binding in plan.capability_bindings() {
533        if !instances.contains_key(binding.provider_instance()) {
534            continue;
535        }
536        let provider = plan
537            .plugin_instance(binding.provider_instance())
538            .expect("validated binding provider should exist");
539        let descriptor = provider
540            .provided_capabilities()
541            .iter()
542            .find(|descriptor| descriptor.capability_id() == binding.capability_id())
543            .expect("validated binding descriptor should exist");
544        if !descriptor.request_operations().is_empty() {
545            let endpoint = instances
546                .get(binding.provider_instance())
547                .and_then(|instance| {
548                    instance.endpoints.request().iter().find(|endpoint| {
549                        endpoint.capability_id() == binding.capability_id()
550                            && endpoint.descriptor_version() == binding.descriptor_version()
551                    })
552                })
553                .cloned()
554                .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
555                    detail: format!(
556                        "Capability `{}` version `{}` has no request endpoint on provider `{}`",
557                        binding.capability_id(),
558                        binding.descriptor_version(),
559                        binding.provider_instance()
560                    ),
561                })?;
562            bindings.push(
563                PreparedBinding::new(
564                    binding.consumer_instance(),
565                    binding.provider_instance(),
566                    endpoint,
567                )
568                .with_requirement_id(binding.requirement_id()),
569            );
570        }
571        if !descriptor.stream_operations().is_empty() {
572            let endpoint = instances
573                .get(binding.provider_instance())
574                .and_then(|instance| {
575                    instance.endpoints.stream().iter().find(|endpoint| {
576                        endpoint.capability_id() == binding.capability_id()
577                            && endpoint.descriptor_version() == binding.descriptor_version()
578                    })
579                })
580                .cloned()
581                .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
582                    detail: format!(
583                        "Capability `{}` version `{}` has no stream endpoint on provider `{}`",
584                        binding.capability_id(),
585                        binding.descriptor_version(),
586                        binding.provider_instance()
587                    ),
588                })?;
589            stream_bindings.push(
590                PreparedStreamBinding::new(
591                    binding.consumer_instance(),
592                    binding.provider_instance(),
593                    endpoint,
594                )
595                .with_requirement_id(binding.requirement_id()),
596            );
597        }
598        if !descriptor.event_operations().is_empty() {
599            let endpoint = instances
600                .get(binding.provider_instance())
601                .and_then(|instance| {
602                    instance.endpoints.event().iter().find(|endpoint| {
603                        endpoint.capability_id() == binding.capability_id()
604                            && endpoint.descriptor_version() == binding.descriptor_version()
605                    })
606                })
607                .cloned()
608                .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
609                    detail: format!(
610                        "Capability `{}` version `{}` has no Event endpoint on provider `{}`",
611                        binding.capability_id(),
612                        binding.descriptor_version(),
613                        binding.provider_instance()
614                    ),
615                })?;
616            event_bindings.push(
617                PreparedEventBinding::new(
618                    binding.consumer_instance(),
619                    binding.provider_instance(),
620                    endpoint,
621                )
622                .with_requirement_id(binding.requirement_id()),
623            );
624        }
625    }
626    Ok((bindings, stream_bindings, event_bindings))
627}
628
629fn invalid<T>(detail: String) -> Result<T, RuntimeFailure> {
630    Err(RuntimeFailure::InvalidResolvedPlan { detail })
631}
lenso_native_adapter/lib.rs

lenso_native_adapter/
lib.rs
