aion/loader/

load.rs

//! Package staging: validated load units shared by the workflow catalog.

use std::collections::HashSet;
use std::time::Duration;

use aion_package::{
    ContentHash, ContractIdentityError, ManifestDigest, ManifestVersion, Package, PackageContract,
    WorkerContract,
};

use crate::error::EngineError;

/// Outcome of one package load, computed inside the catalog mutation lock.
///
/// `freshly_loaded` distinguishes a real registration from an idempotent
/// re-load of a resident hash; `route_changed` reports whether the call
/// re-pointed routing (false means the hash was already route-active and the
/// load was a full no-op). Both flags are race-free truth captured under the
/// same lock that committed the mutation, never a list-before/list-after read.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LoadOutcome {
    /// The loaded (or already-resident) workflow record.
    pub record: LoadedWorkflow,
    /// True when this call registered the version; false on idempotent re-load.
    pub freshly_loaded: bool,
    /// True when this call re-pointed the type's route at the version.
    pub route_changed: bool,
    /// Every OTHER version of this workflow type still resident after the
    /// load, sorted, with the newly routed one excluded.
    ///
    /// Deploy has never removed a superseded version and is not going to
    /// start: a version may still be carrying live runs, and the decision to
    /// unload one is the platform's, not the engine's (decision record #62).
    /// What deploy CAN do is stop the accumulation being invisible. A stale
    /// retained version is still a reachable contract — one of them made a
    /// whole task queue unservable — so the operator is handed the exact
    /// hashes `unload_workflow_version` will accept.
    pub superseded_versions: Vec<ContentHash>,
}

/// One queue contract retained under an exact `.v4` package identity.
///
/// Content-hash namespacing means several versions of the same logical package
/// are retained side by side, so a queue's retained contracts are a SET, not a
/// single current shape. Whether any given one still binds a registering worker
/// depends on reachability, which is why the record carries the routing and
/// membership facts the admission gate decides on rather than the contract
/// alone.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DeployedWorkerContract {
    /// Exact package identity requiring the queue surface.
    pub package_version: ContentHash,
    /// Queue-scoped action declarations from the durable contract record.
    pub contract: WorkerContract,
    /// Every workflow type this exact package version implements, sorted. A
    /// package archive can carry several entry modules, and each is a distinct
    /// catalog entry under the SAME content hash.
    pub workflow_types: Vec<String>,
    /// Whether any of those workflow types currently routes new starts at this
    /// version. A route-active version can be started at any moment, so it
    /// always binds a registering worker.
    pub route_active: bool,
}

/// Workflow package entrypoint registered in the embedded runtime.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LoadedWorkflow {
    workflow_type: String,
    deployed_entry_module: String,
    entry_function: String,
    version: ContentHash,
    declared_timeout: Option<Duration>,
    contract: Option<PackageContract>,
}

impl LoadedWorkflow {
    /// Assembles a loaded-workflow record from already-validated parts.
    ///
    /// `declared_timeout` is the entry's explicitly authored workflow timeout,
    /// or `None` when the package's content-hash identity does not commit to one
    /// (a legacy or defaulted manifest). It is the sole input the start path
    /// consults to decide whether to arm a deadline, so a non-declared entry can
    /// never arm.
    pub(crate) const fn from_parts(
        workflow_type: String,
        deployed_entry_module: String,
        entry_function: String,
        version: ContentHash,
        declared_timeout: Option<Duration>,
        contract: Option<PackageContract>,
    ) -> Self {
        Self {
            workflow_type,
            deployed_entry_module,
            entry_function,
            version,
            declared_timeout,
            contract,
        }
    }

    /// Logical workflow type from the package manifest entry module.
    #[must_use]
    pub fn workflow_type(&self) -> &str {
        &self.workflow_type
    }

    /// Namespaced module name to spawn for this package version.
    #[must_use]
    pub fn deployed_entry_module(&self) -> &str {
        &self.deployed_entry_module
    }

    /// Exported function to spawn for this package version.
    #[must_use]
    pub fn entry_function(&self) -> &str {
        &self.entry_function
    }

    /// Content-hash version identifying this package.
    #[must_use]
    pub fn version(&self) -> &ContentHash {
        &self.version
    }

    /// The entry's explicitly authored workflow timeout, or `None`.
    ///
    /// `Some` only when the package identity commits to a declared timeout; the
    /// start path arms a deadline exactly when this is `Some`, so a legacy or
    /// defaulted manifest — which resolves to `None` here — arms nothing.
    #[must_use]
    pub fn declared_timeout(&self) -> Option<Duration> {
        self.declared_timeout
    }

    /// Returns the durable contract only when this exact identity is `.v4`.
    ///
    /// # Errors
    ///
    /// Returns [`ContractIdentityError::RedeployRequired`] for a pre-`.v4`
    /// package, naming the exact stored identity that must be re-deployed.
    pub fn contract(&self) -> Result<&PackageContract, ContractIdentityError> {
        self.contract
            .as_ref()
            .ok_or_else(|| ContractIdentityError::RedeployRequired {
                stored_version: self.version.to_string(),
            })
    }
}

/// One workflow entry staged from a package manifest.
pub(crate) struct StagedWorkflow {
    pub(crate) workflow_type: String,
    pub(crate) deployed_entry_module: String,
    pub(crate) entry_function: String,
    pub(crate) declared_timeout: Option<Duration>,
}

/// One package validated and decomposed into deployable module units.
pub(crate) struct StagedLoad<'a> {
    pub(crate) workflows: Vec<StagedWorkflow>,
    pub(crate) manifest_version: ManifestVersion,
    pub(crate) manifest_digest: ManifestDigest,
    pub(crate) version: ContentHash,
    pub(crate) modules: Vec<StagedModule<'a>>,
    pub(crate) contract: Option<PackageContract>,
}

impl<'a> StagedLoad<'a> {
    pub(crate) fn new(package: &'a Package) -> Result<Self, EngineError> {
        let manifest = package.manifest();
        let version = package.content_hash().clone();
        let contract = package.contract().ok().cloned();
        // Declaredness is a tamper-evident, authenticated PER-ENTRY property of
        // the content-hash identity: the timeout-bearing identity binds every
        // entry's timeout, so `declared_entry_timeout` returns an entry's authored
        // value only when the identity commits to it. A legacy or defaulted
        // manifest — or one whose additional entries were not bound — reads as
        // wholly undeclared, so each entry's timeout is held non-arming (`None`)
        // regardless of what value its `timeout` field happens to carry.
        let mut seen = HashSet::new();
        let mut workflows = Vec::with_capacity(1 + manifest.additional_workflows.len());
        let entries = std::iter::once((
            manifest.entry_module.as_str(),
            manifest.entry_module.as_str(),
            manifest.entry_function.as_str(),
            manifest.timeout,
        ))
        .chain(manifest.additional_workflows.iter().map(|entry| {
            (
                entry.workflow_type.as_str(),
                entry.entry_module.as_str(),
                entry.entry_function.as_str(),
                entry.timeout,
            )
        }));
        for (workflow_type, entry_module, entry_function, entry_timeout) in entries {
            if !seen.insert(workflow_type) {
                return Err(load_error(format!(
                    "package declares workflow type `{workflow_type}` more than once"
                )));
            }
            if package.beams().get(entry_module).is_none() {
                return Err(load_error(format!(
                    "manifest entry module `{entry_module}` for workflow `{workflow_type}` is absent from package beams"
                )));
            }
            workflows.push(StagedWorkflow {
                workflow_type: workflow_type.to_owned(),
                deployed_entry_module: aion_package::deployed_name(entry_module, &version),
                entry_function: entry_function.to_owned(),
                declared_timeout: package.declared_entry_timeout(entry_timeout),
            });
        }
        let modules = package
            .deployed_modules()
            .into_iter()
            .map(|(deployed_name, bytes)| StagedModule {
                deployed_name,
                bytes,
            })
            .collect();

        Ok(Self {
            workflows,
            manifest_version: manifest.version.clone(),
            manifest_digest: manifest.canonical_digest()?,
            version,
            modules,
            contract,
        })
    }

    /// Loaded-workflow records this package commits atomically.
    pub(crate) fn records(&self) -> Vec<LoadedWorkflow> {
        self.workflows
            .iter()
            .map(|entry| {
                LoadedWorkflow::from_parts(
                    entry.workflow_type.clone(),
                    entry.deployed_entry_module.clone(),
                    entry.entry_function.clone(),
                    self.version.clone(),
                    entry.declared_timeout,
                    self.contract.clone(),
                )
            })
            .collect()
    }
}

/// What the catalog does with a package whose declared contract it cannot
/// enforce.
///
/// The distinction is between a package being OFFERED and a package being
/// RECOVERED, and it is the whole reason this is a parameter rather than a
/// constant: an operator handing over an archive can fix it and must be told
/// now, while a run that has been parked for months cannot fix anything and
/// must not be stranded for a defect in a declaration it never reads.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ContractEnforcement {
    /// Refuse the load. Every door an operator offers a package through:
    /// the deploy seam, and the startup sources a builder is handed.
    Refuse,
    /// Load it, and say so at `error` level. The recovery path only.
    ReportOnly,
}

/// Refuses (or reports) a staged package that declares schemas no validator
/// can compile.
///
/// An uncompilable declared schema does not fail loudly at the boundary it
/// guards: `admit_value` can only answer `UnusableSchema`, and every admission
/// site's answer to that is to let the value through unchecked. So a package
/// carrying one runs with that part of its declared contract switched off, and
/// the only trace is a log line at a moment nobody is watching. The door is
/// the last place it can still be a diagnostic.
///
/// A pre-`.v4` identity commits to no contract at all, so there is nothing to
/// enforce and nothing to report.
pub(crate) fn enforce_contract(
    staged: &StagedLoad<'_>,
    workflow_type: &str,
    enforcement: ContractEnforcement,
) -> Result<(), EngineError> {
    let Some(contract) = staged.contract.as_ref() else {
        return Ok(());
    };
    let unenforceable = contract.unenforceable_schemas();
    if unenforceable.is_empty() {
        return Ok(());
    }
    let detail = unenforceable
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>()
        .join("; ");
    match enforcement {
        ContractEnforcement::Refuse => Err(EngineError::UnenforceableContract {
            workflow_type: workflow_type.to_owned(),
            count: unenforceable.len(),
            detail,
        }),
        ContractEnforcement::ReportOnly => {
            tracing::error!(
                workflow_type,
                version = %staged.version,
                count = unenforceable.len(),
                %detail,
                "recovering a persisted package whose declared contract cannot be enforced: these declarations compile into no validator, so every value they cover is admitted unchecked until the package is re-deployed"
            );
            Ok(())
        }
    }
}

/// One deployable module of a staged package.
pub(crate) struct StagedModule<'a> {
    pub(crate) deployed_name: String,
    pub(crate) bytes: &'a [u8],
}

pub(crate) fn load_error(reason: String) -> EngineError {
    EngineError::Load { reason }
}

/// Best-effort rollback of modules registered before a failed load step.
///
/// Returns a human-readable suffix describing rollback failures, empty when
/// every registration was unwound cleanly.
pub(crate) fn rollback_registered<R>(rollback: &mut R, registered_now: &[String]) -> String
where
    R: FnMut(&str) -> Result<(), EngineError>,
{
    let mut errors = Vec::new();
    for deployed_name in registered_now.iter().rev() {
        if let Err(error) = rollback(deployed_name) {
            errors.push(format!("{deployed_name}: {error}"));
        }
    }

    if errors.is_empty() {
        String::new()
    } else {
        format!("; rollback failed for {}", errors.join(", "))
    }
}