cloacina::registry::loader::task_registrar

Rust

Task registrar for integrating packaged workflow tasks with the global registry.

This module provides functionality to register tasks from dynamically loaded library packages with cloacina’s global task registry, ensuring proper namespace isolation and task lifecycle management.

Re-exports

• OwnedTaskMetadata
• OwnedTaskMetadataCollection

Structs

struct TaskRegistrar

pub

Task registrar for managing dynamically loaded package tasks.

This registrar integrates packaged workflow tasks with cloacina’s global task registry while maintaining proper namespace isolation and lifecycle management for dynamic libraries.

Methods

new

pub

fn new() -> Result<Self, LoaderError>

Create a new task registrar with a temporary directory for operations.

Source

    pub fn new() -> Result<Self, LoaderError> {
        let temp_dir = TempDir::new().map_err(|e| LoaderError::TempDirectory {
            error: e.to_string(),
        })?;

        Ok(Self {
            temp_dir,
            registered_tasks: Arc::new(RwLock::new(HashMap::new())),
            loaded_packages: Arc::new(RwLock::new(HashMap::new())),
            handle_cache: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
        })
    }

with_​handle_​cache

pub

fn with_handle_cache(
    cache: crate::registry::loader::package_loader::PluginHandleCache,
) -> Result<Self, LoaderError>

Create a task registrar with a shared handle cache.

Source

    pub fn with_handle_cache(
        cache: crate::registry::loader::package_loader::PluginHandleCache,
    ) -> Result<Self, LoaderError> {
        let temp_dir = TempDir::new().map_err(|e| LoaderError::TempDirectory {
            error: e.to_string(),
        })?;

        Ok(Self {
            temp_dir,
            registered_tasks: Arc::new(RwLock::new(HashMap::new())),
            loaded_packages: Arc::new(RwLock::new(HashMap::new())),
            handle_cache: cache,
        })
    }

register_​package_​tasks

pub async

async fn register_package_tasks(
    &self,
    package_id: &str,
    package_data: &[u8],
    _metadata: &PackageMetadata,
    tenant_id: Option<&str>,
    runtime: &Arc<Runtime>,
) -> Result<Vec<TaskNamespace>, LoaderError>

Register package tasks with the global task registry using new host-managed approach.

Parameters:

Name	Type	Description
package_id	&str	Unique identifier for the package (for cleanup tracking)
package_data	&[u8]	Binary data of the library package
metadata	-	Package metadata containing task information (legacy, for compatibility)
tenant_id	Option<&str>	Tenant ID for namespace isolation (default: “public”)

Returns:

• Ok(Vec<TaskNamespace>) - List of registered task namespaces
• Err(LoaderError) - If registration fails

Source

    pub async fn register_package_tasks(
        &self,
        package_id: &str,
        package_data: &[u8],
        _metadata: &PackageMetadata,
        tenant_id: Option<&str>,
        runtime: &Arc<Runtime>,
    ) -> Result<Vec<TaskNamespace>, LoaderError> {
        let tenant_id = tenant_id.unwrap_or("public");

        // Extract task metadata from library using new FFI approach.
        // This returns owned data - all strings are copied before the library is unloaded.
        let task_metadata = self
            .extract_task_metadata_from_library(package_data)
            .await?;

        // Load the plugin library once — all tasks from this package share the handle.
        let plugin = Arc::new(
            DynamicLibraryTask::load_plugin(package_data, &task_metadata.package_name).map_err(
                |e| LoaderError::MetadataExtraction {
                    reason: format!("Failed to load plugin for task execution: {}", e),
                },
            )?,
        );

        // Register tasks in HOST global registry using metadata.
        let mut registered_namespaces = Vec::new();

        let workflow_name = &task_metadata.workflow_name;
        let package_name = &task_metadata.package_name;

        for task in &task_metadata.tasks {
            let task_id = &task.local_id;
            let dependencies_json = &task.dependencies_json;

            // Parse dependencies JSON to get dependency namespaces
            let dependency_namespaces: Vec<TaskNamespace> = if dependencies_json.trim() == "[]" {
                Vec::new()
            } else {
                let dep_names: Vec<String> =
                    serde_json::from_str(dependencies_json).map_err(|e| {
                        LoaderError::MetadataExtraction {
                            reason: format!(
                                "Failed to parse dependencies JSON '{}': {}",
                                dependencies_json, e
                            ),
                        }
                    })?;

                dep_names
                    .into_iter()
                    .map(|dep_name| {
                        if dep_name.contains("::") {
                            let full_name = dep_name.replace("{tenant}", tenant_id);
                            crate::parse_namespace(&full_name).map_err(|e| {
                                LoaderError::MetadataExtraction {
                                    reason: format!(
                                        "Invalid dependency namespace '{}': {}",
                                        full_name, e
                                    ),
                                }
                            })
                        } else {
                            Ok(TaskNamespace::new(
                                tenant_id,
                                package_name,
                                workflow_name,
                                &dep_name,
                            ))
                        }
                    })
                    .collect::<Result<Vec<_>, _>>()?
            };

            let namespace = TaskNamespace::new(tenant_id, package_name, workflow_name, task_id);

            // Trigger rules for conditional execution (CLOACI-T-0721) — parsed
            // from the cdylib's FFI metadata so packaged workflows honour skips.
            // Falls back to Always on any parse issue.
            let trigger_rules: serde_json::Value = serde_json::from_str(&task.trigger_rules_json)
                .unwrap_or_else(|_| serde_json::json!({ "type": "Always" }));

            let plugin = plugin.clone();
            let task_name = task_id.to_string();
            let deps = dependency_namespaces.clone();

            // CLOACI-T-0897: carry the handle requirement through to the task,
            // or the executor skips the handle path and defer_until cannot work.
            let requires_handle = task.requires_handle;

            let constructor = Box::new(move || {
                Arc::new(DynamicLibraryTask::new(
                    plugin.clone(),
                    task_name.clone(),
                    deps.clone(),
                    trigger_rules.clone(),
                    requires_handle,
                )) as Arc<dyn Task>
            });

            runtime.register_task(namespace.clone(), constructor);

            registered_namespaces.push(namespace);
        }

        // Track registered tasks for cleanup
        {
            let mut registered = self.registered_tasks.write();
            registered.insert(package_id.to_string(), registered_namespaces.clone());
        }

        tracing::info!(
            "Successfully registered {} tasks for package {} using host-managed approach",
            registered_namespaces.len(),
            package_name
        );

        Ok(registered_namespaces)
    }

unregister_​package_​tasks

pub

fn unregister_package_tasks(
    &self) -> Result<(,
    package_id: &str) -> Result<(,
) -> Result<(), LoaderError>

Unregister package tasks from the global registry.

Parameters:

Name	Type	Description
package_id	&str	Package identifier used during registration

Returns:

• Ok(()) - Tasks successfully unregistered
• Err(LoaderError) - If unregistration fails

Source

    pub fn unregister_package_tasks(&self, package_id: &str) -> Result<(), LoaderError> {
        // Remove from tracked registrations
        let namespaces = {
            let mut registered = self.registered_tasks.write();
            registered.remove(package_id)
        };

        if let Some(namespaces) = namespaces {
            // Unregister tasks from global registry
            // Note: The global registry doesn't currently support removal,
            // so we'll track this for future implementation
            tracing::warn!(
                "Task unregistration requested for package '{}' with {} tasks, but global registry doesn't support removal yet",
                package_id,
                namespaces.len()
            );
        }

        // Remove package tracking entry
        {
            let mut packages = self.loaded_packages.write();
            packages.remove(package_id);
        }

        Ok(())
    }

get_​registered_​namespaces

pub

fn get_registered_namespaces(
    &self,
    package_id: &str,
) -> Vec<TaskNamespace>

Get the list of task namespaces registered for a package.

Source

    pub fn get_registered_namespaces(&self, package_id: &str) -> Vec<TaskNamespace> {
        let registered = self.registered_tasks.read();
        registered.get(package_id).cloned().unwrap_or_default()
    }

loaded_​package_​count

pub

fn loaded_package_count(&self) -> usize

Get the number of currently loaded packages.

Source

    pub fn loaded_package_count(&self) -> usize {
        let packages = self.loaded_packages.read();
        packages.len()
    }

total_​registered_​tasks

pub

fn total_registered_tasks(&self) -> usize

Get the total number of registered tasks across all packages.

Source

    pub fn total_registered_tasks(&self) -> usize {
        let registered = self.registered_tasks.read();
        registered.values().map(|tasks| tasks.len()).sum()
    }

temp_​dir

pub

fn temp_dir(&self) -> &Path

Get the temporary directory path for manual operations.

Source

    pub fn temp_dir(&self) -> &Path {
        self.temp_dir.path()
    }