aion-rs 0.13.2

//! `Engine` runtime package-load seam: live load, routing, listing, unload.
//!
//! Decision record (#62, adopted 2026-06-12): D1 always-latest-at-record-time
//! with durable pinning, D2 manual unload with engine-enforced safety checks,
//! D3 embedded-only API with serde-ready types (the server endpoint is a
//! follow-up brief), D4 required `package_version` on start events.

use aion_core::Event;
use aion_package::ContentHash;

use crate::error::PinHolder;
use crate::loader::{DeclaredQueues, DeployedWorkerContract, LoadOutcome, WorkflowVersionInfo};
use crate::{EngineError, WorkflowCatalog};

use super::api::Engine;
use super::builder::WorkflowPackageSource;
use super::builder_assembly::package_from_source;

impl Engine {
    /// Loads a validated package into the running engine and atomically
    /// routes its workflow type's new dispatches to it.
    ///
    /// Every start that resolved before the route flip completes on the
    /// version it resolved (loads never unregister anything); every start
    /// after this call returns resolves the new version. Re-loading an
    /// already-loaded hash is idempotent (nothing registers,
    /// `freshly_loaded = false`) but still re-points the route at it —
    /// re-deploying a previously rolled-back version must take effect
    /// (`route_changed` reports whether it did).
    ///
    /// Verified modules and catalog entries are staged first, then the archive
    /// and route are persisted, and only then are in-memory routes published.
    /// Therefore no start can record a hash whose archive is not durable.
    /// startup reloads every persisted package before recovery resolves any
    /// run's recorded pinned version. Idempotent re-loads re-persist —
    /// re-deploying is a routing intent and the durable pointer must mirror
    /// it.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::ShuttingDown`] once shutdown begins,
    /// [`EngineError::Load`] for archive, collision, registration, or
    /// entry-verification failures, and [`EngineError::ManifestMismatch`]
    /// when an idempotent re-load presents the resident content hash with a
    /// different manifest. On those failures live routing is untouched:
    /// routing, loaded versions, and in-flight dispatches are unaffected.
    /// Returns [`EngineError::Store`] when persistence fails; newly staged
    /// modules and entries are rolled back before the error is returned.
    pub async fn load_package(
        &self,
        source: impl Into<WorkflowPackageSource>,
    ) -> Result<LoadOutcome, EngineError> {
        // A load is new-work admission, not a wind-down operation: refuse
        // after shutdown begins so modules never register into a dying VM.
        let operation = self.shutdown_gate.begin_start()?;
        let result = async {
            // Catalog commit and persistence are one deploy mutation: an
            // interleaved route/unload/deploy between them could persist
            // state the catalog no longer holds.
            let _deploy = self.deploy_mutations.lock().await;
            let package = package_from_source(source.into())?;
            let catalog = self.workflow_catalog();
            let outcome = catalog.stage_package(self.runtime(), &package).await?;
            let version = package.content_hash().clone();
            if let Err(error) = crate::loader::persistence::persist_deployed_package(
                self.store().as_ref(),
                &package,
            )
            .await
            {
                if outcome.freshly_loaded {
                    let mutation_guard = catalog.begin_mutation().await;
                    let removed = catalog
                        .swap_out_package(package.manifest().entry_module.as_str(), &version)?;
                    self.unregister_unloaded_modules(
                        package.manifest().entry_module.as_str(),
                        &version,
                        &removed,
                    )?;
                    drop(mutation_guard);
                }
                return Err(error);
            }
            catalog
                .publish_package_routes(package.manifest().entry_module.as_str(), &version)
                .await?;
            // Read the superseded set AFTER routing settles, so the version
            // this call just routed is never in it.
            let mut outcome = outcome;
            outcome.superseded_versions =
                catalog.superseded_versions(outcome.record.workflow_type())?;
            Ok(outcome)
        }
        .await;
        drop(operation);
        result
    }

    /// Lists every loaded workflow version with its routing flag, sorted by
    /// `(workflow_type, loaded_at)`.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::CatalogPoisoned`] when the catalog lock is poisoned.
    pub fn list_workflow_versions(&self) -> Result<Vec<WorkflowVersionInfo>, EngineError> {
        self.workflow_catalog().versions()
    }

    /// Returns every retained `.v4` package contract declaring `task_queue`.
    ///
    /// This is the RAW retained set, which under content-hash namespacing holds
    /// every coexisting version — including ones nothing can reach any more.
    /// Worker admission must not be decided from it directly; use
    /// [`Engine::worker_contracts_for_admission`], which splits it into the
    /// reachable versions that bind a connection and the unreachable ones that
    /// bind nobody. Demanding the raw set of one connection is what made a
    /// queue with a single stale version permanently unservable.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::CatalogPoisoned`] when the catalog lock is poisoned.
    pub fn worker_contracts_for_queue(
        &self,
        task_queue: &str,
    ) -> Result<Vec<DeployedWorkerContract>, EngineError> {
        self.workflow_catalog()
            .worker_contracts_for_queue(task_queue)
    }

    /// Returns one read of the task queues the retained contracts declare.
    ///
    /// The server's queue-service classifier (R1) reads this to tell a
    /// structurally undeclared queue apart from a declared but unserved one.
    /// The answer reports whether it covered every retained entry: a queue
    /// missing from a read that could not decode some entry is unknowable, not
    /// undeclared. An empty set likewise means this catalog declares no queues
    /// at all and therefore cannot contradict any dispatch — see
    /// [`WorkflowCatalog::declared_task_queues`](crate::loader::WorkflowCatalog::declared_task_queues).
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::CatalogPoisoned`] when the catalog lock is poisoned.
    pub fn declared_task_queues(&self) -> Result<DeclaredQueues, EngineError> {
        self.workflow_catalog().declared_task_queues()
    }

    /// Re-points routing for `workflow_type` at an already-loaded version
    /// (rollback / roll-forward). Atomic and idempotent.
    ///
    /// The pointer is persisted so the re-point survives a restart; startup
    /// restores persisted pointers after reloading persisted packages.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::ShuttingDown`] once shutdown begins,
    /// [`EngineError::UnknownVersion`] naming the loaded set when
    /// `(type, version)` is not loaded — routing to a never-loaded hash is
    /// impossible — and [`EngineError::Store`] when the durable pointer could
    /// not be written. The in-memory route is published only after that write.
    pub async fn route_workflow_version(
        &self,
        workflow_type: &str,
        version: &ContentHash,
    ) -> Result<(), EngineError> {
        let operation = self.shutdown_gate.begin_operation()?;
        let result = async {
            // One deploy mutation: the catalog re-point and the durable
            // pointer write must not interleave with another deploy
            // mutation's persistence.
            let _deploy = self.deploy_mutations.lock().await;
            let catalog = self.workflow_catalog();
            if catalog.get(workflow_type, version)?.is_none() {
                let loaded_versions = catalog
                    .versions()?
                    .into_iter()
                    .filter(|entry| entry.workflow_type == workflow_type)
                    .map(|entry| entry.content_hash.to_string())
                    .collect::<Vec<_>>();
                let loaded = if loaded_versions.is_empty() {
                    "none".to_owned()
                } else {
                    loaded_versions.join(", ")
                };
                return Err(EngineError::UnknownVersion {
                    workflow_type: workflow_type.to_owned(),
                    version: version.clone(),
                    loaded,
                });
            }
            self.store()
                .put_package_route(workflow_type, &version.to_string())
                .await?;
            catalog.route_version(workflow_type, version).await?;
            Ok(())
        }
        .await;
        drop(operation);
        result
    }

    /// Unloads a workflow version after verifying nothing pins it (D2).
    ///
    /// Refusal conditions, each typed and naming what pins the version:
    /// route-inactive is required (the route-active version of a type can
    /// never be unloaded), no in-flight start may pin it, no live registry
    /// handle may run on it, and no recoverable instance in the store —
    /// running, durably paused, or a recorded-but-never-started child — may be
    /// pinned to it.
    ///
    /// The engine owns the mechanism; the embedding platform owns *when* to
    /// unload. There is no automatic garbage collection.
    ///
    /// Unload deletes the persisted deploy artifact too (a no-op for
    /// versions loaded from operator files, which were never persisted), so
    /// an unloaded version does not resurrect at the next restart.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::ShuttingDown`] once shutdown begins,
    /// [`EngineError::UnknownVersion`] when `(type, version)` is not loaded,
    /// [`EngineError::RouteActive`] when the version is route-active,
    /// [`EngineError::VersionPinned`] naming the concrete pin holder (with
    /// the catalog restored untouched), [`EngineError::Store`] when the
    /// persisted artifact could not be deleted (the catalog is restored and
    /// the unload did not happen), and [`EngineError::Runtime`] when module
    /// unregistration fails after the catalog commit.
    pub async fn unload_workflow_version(
        &self,
        workflow_type: &str,
        version: &ContentHash,
    ) -> Result<(), EngineError> {
        let operation = self.shutdown_gate.begin_operation()?;
        let result = self
            .unload_workflow_version_inner(workflow_type, version)
            .await;
        drop(operation);
        result
    }

    async fn unload_workflow_version_inner(
        &self,
        workflow_type: &str,
        version: &ContentHash,
    ) -> Result<(), EngineError> {
        let catalog = self.workflow_catalog();
        // Deploy lock first (the engine-wide ordering: deploy_mutations,
        // then the catalog mutation lock), so the persisted-artifact delete
        // below cannot interleave with a concurrent re-deploy's persistence.
        let _deploy = self.deploy_mutations.lock().await;
        let _mutation = catalog.begin_mutation().await;
        // Swap the version out FIRST: from this instant no new resolution can
        // produce it, so the checks below cannot be invalidated by a racing
        // start (a start that already resolved holds a pin and is detected).
        let removed = catalog.swap_out_package(workflow_type, version)?;
        let member_types = removed
            .workflow_types()
            .map(str::to_owned)
            .collect::<Vec<_>>();
        if let Err(error) = self
            .verify_unload_unpinned(catalog, &member_types, version)
            .await
        {
            catalog.restore_package(removed)?;
            return Err(error);
        }
        // Delete the persisted artifact BEFORE unregistering modules: if the
        // delete fails the unload is rolled back wholesale, never leaving a
        // version that is gone from this process yet resurrects at the next
        // restart. Idempotent for never-persisted (operator-file) versions.
        if let Err(error) = self
            .store()
            .delete_package(removed.primary_workflow_type(), &version.to_string())
            .await
        {
            catalog.restore_package(removed)?;
            return Err(error.into());
        }
        self.unregister_unloaded_modules(workflow_type, version, &removed)
    }

    /// Verifies no member type in an archive group is pinned to `version`.
    async fn verify_unload_unpinned(
        &self,
        catalog: &WorkflowCatalog,
        workflow_types: &[String],
        version: &ContentHash,
    ) -> Result<(), EngineError> {
        for workflow_type in workflow_types {
            self.verify_unload_member_unpinned(catalog, workflow_type, version)
                .await?;
        }
        Ok(())
    }

    async fn verify_unload_member_unpinned(
        &self,
        catalog: &WorkflowCatalog,
        workflow_type: &str,
        version: &ContentHash,
    ) -> Result<(), EngineError> {
        if catalog.has_pinned_starts(workflow_type, version)? {
            return Err(EngineError::VersionPinned {
                workflow_type: workflow_type.to_owned(),
                version: version.clone(),
                pinned_by: PinHolder::InFlightStart,
            });
        }

        for handle in self.registry().list()? {
            if handle.workflow_type() == workflow_type
                && handle.loaded_version() == version
                && !handle.cached_status().is_terminal()
            {
                return Err(EngineError::VersionPinned {
                    workflow_type: workflow_type.to_owned(),
                    version: version.clone(),
                    pinned_by: PinHolder::LiveRun {
                        workflow_id: handle.workflow_id().clone(),
                        run_id: handle.run_id().clone(),
                    },
                });
            }
        }

        let recorded = crate::loader::package_version_of(version);
        let store = self.store();
        // `list_active` projects `Running` ONLY — a durably-`Paused` run is
        // non-terminal, is not respawned by startup recovery (so it holds no
        // registry handle either), and an operator `resume` puts it straight
        // back on this exact version. Scanning the active set alone therefore
        // let an unload delete the module a paused run would resume into,
        // stranding it permanently. Both projections are scanned; a workflow in
        // both sets is simply checked twice.
        let mut recoverable = store.list_active().await?;
        recoverable.extend(store.list_paused().await?);
        for workflow_id in recoverable {
            let history = store.read_history(&workflow_id).await?;
            let current_run_pin = history.iter().rev().find_map(|event| match event {
                Event::WorkflowStarted {
                    workflow_type: started_type,
                    package_version,
                    ..
                } => Some(started_type == workflow_type && package_version == &recorded),
                _ => None,
            });
            if current_run_pin == Some(true) {
                return Err(EngineError::VersionPinned {
                    workflow_type: workflow_type.to_owned(),
                    version: version.clone(),
                    pinned_by: PinHolder::RecoverableRun {
                        workflow_id: workflow_id.clone(),
                    },
                });
            }
            if let Some(child_workflow_id) = self
                .recorded_unstarted_child_pin(&history, workflow_type, &recorded)
                .await?
            {
                return Err(EngineError::VersionPinned {
                    workflow_type: workflow_type.to_owned(),
                    version: version.clone(),
                    pinned_by: PinHolder::RecordedChild {
                        child_workflow_id,
                        recorded_by: workflow_id.clone(),
                    },
                });
            }
        }
        Ok(())
    }

    /// A recorded-but-never-started child pinned to the target version, if
    /// any: its `ChildWorkflowStarted` carries the version and its own
    /// history is still empty, so the crash-repair sweep would have to start
    /// it on exactly this version.
    async fn recorded_unstarted_child_pin(
        &self,
        parent_history: &[Event],
        workflow_type: &str,
        recorded: &aion_core::PackageVersion,
    ) -> Result<Option<aion_core::WorkflowId>, EngineError> {
        let store = self.store();
        for event in parent_history {
            let Event::ChildWorkflowStarted {
                child_workflow_id,
                workflow_type: child_type,
                package_version,
                ..
            } = event
            else {
                continue;
            };
            if child_type != workflow_type || package_version != recorded {
                continue;
            }
            if store.read_history(child_workflow_id).await?.is_empty() {
                return Ok(Some(child_workflow_id.clone()));
            }
        }
        Ok(None)
    }

    /// Unregisters the removed version's modules from the runtime, skipping
    /// host NIF modules that were never BEAM-registered.
    fn unregister_unloaded_modules(
        &self,
        workflow_type: &str,
        version: &ContentHash,
        removed: &crate::loader::catalog::RemovedPackage,
    ) -> Result<(), EngineError> {
        let nif_modules = self.runtime().registered_nif_modules();
        let mut failures = Vec::new();
        for deployed_name in removed.module_names() {
            let original = deployed_name.split('$').next().unwrap_or(deployed_name);
            if nif_modules.iter().any(|name| name == original) {
                continue;
            }
            if let Err(error) = self.runtime().unregister_module(deployed_name) {
                failures.push(format!("{deployed_name}: {error}"));
            }
        }
        if failures.is_empty() {
            Ok(())
        } else {
            // The catalog commit stands: the version is unloaded and its
            // names are unreachable (content-hash unique), but the runtime
            // retains orphaned module entries.
            Err(EngineError::Runtime {
                reason: format!(
                    "workflow `{workflow_type}` version `{version}` was removed from the catalog but module unregistration failed for {}",
                    failures.join(", ")
                ),
            })
        }
    }
}