Packaged Workflow Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                        Cloacina Application                         │
├─────────────────────────────────────────────────────────────────────┤
│                          DefaultRunner                              │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐     │
│  │   Task          │  │   Pipeline      │  │   Background    │     │
│  │   Scheduler     │  │   Executor      │  │   Services      │     │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘     │
├─────────────────────────────────────────────────────────────────────┤
│                      Global Task Registry                           │
│         (Unified namespace for embedded + packaged tasks)           │
├─────────────────────────────────────────────────────────────────────┤
│  Embedded Workflows        │        Workflow Packages                │
│  ┌─────────────────┐      │      ┌─────────────────┐                │
│  │ #[workflow]     │      │      │ Workflow        │                │
│  │ #[task] macro   │      │      │ Registry        │                │
│  │ Compile-time    │      │      │ - Storage       │                │
│  │ registration    │      │      │ - Loader        │                │
│  └─────────────────┘      │      │ - Validator     │                │
│                           │      │ - Reconciler    │                │
│                           │      └─────────────────┘                │
├─────────────────────────────────────────────────────────────────────┤
│                        Database Layer                               │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐     │
│  │ Execution State │  │ Package         │  │ Cron Schedules  │     │
│  │ - Pipelines     │  │ Metadata        │  │ - Time-based    │     │
│  │ - Tasks         │  │ - Binary Store  │  │ - Recovery      │     │
│  │ - Context       │  │ - Registry      │  │ - Missed runs   │     │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘     │
└─────────────────────────────────────────────────────────────────────┘

pub struct DefaultRunner {
    /// Database connection for persistence and state management
    database: Database,
    /// Configuration parameters for the runner
    config: DefaultRunnerConfig,
    /// Task scheduler for managing workflow execution scheduling
    scheduler: Arc<TaskScheduler>,
    /// Task executor for running individual tasks
    executor: Arc<dyn TaskExecutorTrait>,
    /// Optional workflow registry for workflow packages
    workflow_registry: Arc<RwLock<Option<Arc<WorkflowRegistryImpl<FilesystemRegistryStorage>>>>>,
    /// Optional registry reconciler for workflow packages
    registry_reconciler: Arc<RwLock<Option<Arc<RegistryReconciler>>>>,
    /// Optional cron scheduler for time-based workflow execution
    cron_scheduler: Arc<RwLock<Option<Arc<CronScheduler>>>>,
}

pub struct WorkflowRegistryImpl<S: RegistryStorage> {
    /// Storage backend for binary data (.cloacina files)
    storage: S,
    /// Database for metadata storage
    database: Database,
    /// Package loader for metadata extraction
    loader: PackageLoader,
    /// Task registrar — projects packaged tasks into the host's `Runtime`
    /// task registry so they're discoverable alongside embedded tasks.
    registrar: TaskRegistrar,
    /// Package validator for safety checks
    validator: PackageValidator,
    /// Map of package IDs to registered task namespaces
    loaded_packages: HashMap<Uuid, Vec<TaskNamespace>>,
}

// Emitted by the #[task] macro for each task in a #[workflow] module.
// `Runtime::seed_from_inventory()` walks these entries at startup.
cloacina_workflow_plugin::inventory::submit! {
    cloacina_workflow_plugin::TaskEntry {
        namespace: || cloacina::TaskNamespace::new(
            "public", "embedded", "embedded_workflow", "collect_data",
        ),
        constructor: || std::sync::Arc::new(CollectDataTask::new()),
    }
}

// Runtime registration via TaskRegistrar
registrar.register_package_tasks(
    &package_metadata,
    &task_namespaces,
    library_handle
).await?;

// Embedded workflows only
let runner = DefaultRunner::new(&database_url).await?;

// With workflow package support
let mut config = DefaultRunnerConfig::default();
config.enable_registry_reconciler = true;
config.registry_storage_path = Some(PathBuf::from("/path/to/storage"));

let runner = DefaultRunner::with_config(&database_url, config).await?;

CREATE TABLE workflow_packages (
    id UUID PRIMARY KEY,
    package_name TEXT NOT NULL,
    version TEXT NOT NULL,
    description TEXT,
    author TEXT,
    created_at TIMESTAMPTZ NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL,
    UNIQUE(package_name, version)
);

CREATE TABLE workflow_registry (
    id UUID PRIMARY KEY,
    package_id UUID NOT NULL REFERENCES workflow_packages(id),
    binary_data BYTEA NOT NULL,
    created_at TIMESTAMPTZ NOT NULL
);

// Same execution interface for both types
impl DefaultRunner {
    pub async fn execute<T>(&self, workflow_name: &str, context: Context<T>) -> Result<()>
    where T: Clone + Send + Sync + 'static {
        // 1. Create pipeline execution record
        let pipeline_id = self.pipeline_executor.create_pipeline(workflow_name, context).await?;

        // 2. Task discovery (embedded or packaged)
        let tasks = self.scheduler.get_workflow_tasks(workflow_name).await?;

        // 3. Dependency resolution and scheduling
        let execution_plan = self.scheduler.create_execution_plan(tasks).await?;

        // 4. Execute tasks (same executor for both types)
        self.executor.execute_plan(pipeline_id, execution_plan).await?;

        Ok(())
    }
}

pub struct RegistryReconciler {
    registry: Arc<WorkflowRegistryImpl<FilesystemRegistryStorage>>,
    config: ReconcilerConfig,
}

impl RegistryReconciler {
    pub async fn run(&self, mut shutdown: broadcast::Receiver<()>) {
        loop {
            // Scan storage for new/updated packages
            self.scan_and_reconcile().await;

            // Wait for next poll interval or shutdown
            tokio::select! {
                _ = shutdown.recv() => break,
                _ = tokio::time::sleep(self.config.poll_interval) => continue,
            }
        }
    }
}

// Tenant A
let runner_a = DefaultRunnerBuilder::new()
    .database_url("postgresql://user:pass@localhost/cloacina")
    .schema("tenant_a")
    .build()
    .await?;

// Tenant B
let runner_b = DefaultRunnerBuilder::new()
    .database_url("postgresql://user:pass@localhost/cloacina")
    .schema("tenant_b")
    .build()
    .await?;

// Tenant-specific storage paths
config.registry_storage_path = Some(PathBuf::from("/storage/tenant_a"));

crates/
  cloacina/                      # Full runtime: executor, scheduler, DAL, runner, registry, multi-tenancy
  cloacina-workflow/             # Slim author-facing types (Trigger trait, RetryPolicy, error types)
  cloacina-workflow-plugin/      # FFI vtable trait shared by host + plugin (the nine-method CloacinaPlugin)
  cloacina-computation-graph/    # Computation-graph runtime: reactors, accumulators, scheduler
  cloacina-macros/               # Procedural macros: #[task], #[workflow], #[trigger], #[reactor], #[computation_graph], #[accumulator]
  cloacina-build/                # Build-script helper for cdylib linker flags / rpath
  cloacina-compiler/             # Standalone build worker (cargo build for .cloacina packages)
  cloacina-server/               # HTTP API server binary
  cloacina-python/               # PyO3 wheel runtime (the `cloaca` Python module) — isolated per T-0529 so non-Python binaries don't transitively link pyo3
  cloacina-testing/              # Test fixtures: TestRunner, BoundaryEmitter
  cloacinactl/                   # CLI binary (noun-verb command structure per I-0098)

// Production configuration
let mut config = DefaultRunnerConfig::default();
config.max_concurrent_tasks = 50;
config.task_timeout = Duration::from_mins(30);
config.enable_registry_reconciler = true;
config.enable_cron_scheduling = true;
config.enable_recovery = true;

let runner = DefaultRunner::with_config(&database_url, config).await?;

gdoc_arrow_left_alt FFI System Guaranteed Cron Scheduling gdoc_arrow_right_alt