aion/loader/
load.rs

1//! Package staging: validated load units shared by the workflow catalog.
2
3use aion_package::{ContentHash, ManifestDigest, ManifestVersion, Package};
4
5use crate::error::EngineError;
6
7/// Outcome of one package load, computed inside the catalog mutation lock.
8///
9/// `freshly_loaded` distinguishes a real registration from an idempotent
10/// re-load of a resident hash; `route_changed` reports whether the call
11/// re-pointed routing (false means the hash was already route-active and the
12/// load was a full no-op). Both flags are race-free truth captured under the
13/// same lock that committed the mutation, never a list-before/list-after read.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub struct LoadOutcome {
16    /// The loaded (or already-resident) workflow record.
17    pub record: LoadedWorkflow,
18    /// True when this call registered the version; false on idempotent re-load.
19    pub freshly_loaded: bool,
20    /// True when this call re-pointed the type's route at the version.
21    pub route_changed: bool,
22}
23
24/// Workflow package entrypoint registered in the embedded runtime.
25#[derive(Clone, Debug, Eq, PartialEq)]
26pub struct LoadedWorkflow {
27    workflow_type: String,
28    deployed_entry_module: String,
29    entry_function: String,
30    version: ContentHash,
31}
32
33impl LoadedWorkflow {
34    /// Assembles a loaded-workflow record from already-validated parts.
35    pub(crate) const fn from_parts(
36        workflow_type: String,
37        deployed_entry_module: String,
38        entry_function: String,
39        version: ContentHash,
40    ) -> Self {
41        Self {
42            workflow_type,
43            deployed_entry_module,
44            entry_function,
45            version,
46        }
47    }
48
49    /// Logical workflow type from the package manifest entry module.
50    #[must_use]
51    pub fn workflow_type(&self) -> &str {
52        &self.workflow_type
53    }
54
55    /// Namespaced module name to spawn for this package version.
56    #[must_use]
57    pub fn deployed_entry_module(&self) -> &str {
58        &self.deployed_entry_module
59    }
60
61    /// Exported function to spawn for this package version.
62    #[must_use]
63    pub fn entry_function(&self) -> &str {
64        &self.entry_function
65    }
66
67    /// Content-hash version identifying this package.
68    #[must_use]
69    pub fn version(&self) -> &ContentHash {
70        &self.version
71    }
72}
73
74/// One package validated and decomposed into deployable module units.
75pub(crate) struct StagedLoad<'a> {
76    pub(crate) workflow_type: String,
77    pub(crate) deployed_entry_module: String,
78    pub(crate) entry_function: String,
79    pub(crate) manifest_version: ManifestVersion,
80    pub(crate) manifest_digest: ManifestDigest,
81    pub(crate) version: ContentHash,
82    pub(crate) modules: Vec<StagedModule<'a>>,
83}
84
85impl<'a> StagedLoad<'a> {
86    pub(crate) fn new(package: &'a Package) -> Result<Self, EngineError> {
87        let manifest = package.manifest();
88        if package.beams().get(&manifest.entry_module).is_none() {
89            return Err(load_error(format!(
90                "manifest entry module `{}` is absent from package beams",
91                manifest.entry_module
92            )));
93        }
94
95        let version = package.content_hash().clone();
96        let modules = package
97            .deployed_modules()
98            .into_iter()
99            .map(|(deployed_name, bytes)| StagedModule {
100                deployed_name,
101                bytes,
102            })
103            .collect();
104
105        Ok(Self {
106            workflow_type: manifest.entry_module.clone(),
107            deployed_entry_module: package.deployed_entry_module(),
108            entry_function: manifest.entry_function.clone(),
109            manifest_version: manifest.version.clone(),
110            manifest_digest: manifest.canonical_digest()?,
111            version,
112            modules,
113        })
114    }
115
116    /// Loaded-workflow record this staged unit commits as.
117    pub(crate) fn record(&self) -> LoadedWorkflow {
118        LoadedWorkflow::from_parts(
119            self.workflow_type.clone(),
120            self.deployed_entry_module.clone(),
121            self.entry_function.clone(),
122            self.version.clone(),
123        )
124    }
125}
126
127/// One deployable module of a staged package.
128pub(crate) struct StagedModule<'a> {
129    pub(crate) deployed_name: String,
130    pub(crate) bytes: &'a [u8],
131}
132
133pub(crate) fn load_error(reason: String) -> EngineError {
134    EngineError::Load { reason }
135}
136
137/// Best-effort rollback of modules registered before a failed load step.
138///
139/// Returns a human-readable suffix describing rollback failures, empty when
140/// every registration was unwound cleanly.
141pub(crate) fn rollback_registered<R>(rollback: &mut R, registered_now: &[String]) -> String
142where
143    R: FnMut(&str) -> Result<(), EngineError>,
144{
145    let mut errors = Vec::new();
146    for deployed_name in registered_now.iter().rev() {
147        if let Err(error) = rollback(deployed_name) {
148            errors.push(format!("{deployed_name}: {error}"));
149        }
150    }
151
152    if errors.is_empty() {
153        String::new()
154    } else {
155        format!("; rollback failed for {}", errors.join(", "))
156    }
157}
aion/loader/load.rs

aion/loader/
load.rs
