harmos_domain/inventory/
declaration.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::time::Duration;
18// Third Party Imports
19use semver::Version;
20use serde::Deserialize;
21// Local Imports
22use super::contributions::{self, Section};
23use super::{contracts, listeners, observers, publications};
24use crate::operations::{Contract, Definition};
25use crate::{FRAGMENT_LIMIT, ScanError, StreamDeclaration, artifact_subset};
26
27/* === Constants === */
28
29const OWNED: [&str; 14] = [
30    "capabilities",
31    "contracts",
32    "contributions",
33    "guest",
34    "listeners",
35    "observers",
36    "operations",
37    "plugin",
38    "publications",
39    "requires",
40    "restart",
41    "schema",
42    "sidecar",
43    "subscribe",
44];
45
46/* === Macros === */
47
48/* === Type Aliases === */
49
50/* === Structs === */
51
52#[derive(Clone, Debug, PartialEq, Eq)]
53pub struct Declaration {
54    pub id: String,
55    pub version: Version,
56    pub description: String,
57    pub config: String,
58    pub operations: Vec<Definition>,
59    pub contracts: contracts::Manifest,
60    pub listeners: Vec<crate::listeners::Definition>,
61    pub streams: Vec<StreamDeclaration>,
62    pub observers: Vec<observers::Declaration>,
63    pub sections: Vec<Section>,
64    pub carrier: CarrierSection,
65}
66
67#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
68pub struct RestartPolicy {
69    pub credit: u32,
70    pub sustained_health_ms: u64,
71    pub backoff_base_ms: u64,
72    pub backoff_cap_ms: u64,
73}
74
75/* === Enums === */
76
77#[derive(Clone, Copy, Debug, PartialEq, Eq)]
78pub enum Facility {
79    Sidecar,
80    Guest,
81}
82
83#[derive(Clone, Debug, PartialEq, Eq)]
84pub enum CarrierSection {
85    Sidecar { restart: Option<RestartPolicy> },
86    Guest,
87}
88
89#[derive(Debug, thiserror::Error)]
90pub enum DeclarationError {
91    #[error("the artifact uses another Harmos metadata format; rebuild it with the current SDK")]
92    IncompatiblePayload,
93
94    #[error(transparent)]
95    Schema(#[from] crate::schema::graph::Error),
96
97    #[error(transparent)]
98    PublicationInventory(#[from] publications::Error),
99
100    #[error(transparent)]
101    ContractInventory(#[from] contracts::Error),
102
103    #[error(transparent)]
104    ContributionInventory(#[from] contributions::Error),
105
106    #[error(transparent)]
107    ListenerInventory(#[from] listeners::Error),
108
109    #[error(transparent)]
110    ObserverInventory(#[from] observers::Error),
111
112    #[error("guest observer callbacks are unsupported")]
113    UnsupportedObservers,
114
115    #[error("the Harmos sidecar restart policy is invalid")]
116    InvalidRestartPolicy,
117
118    #[error("the artifact carries no Harmos {facility} declaration")]
119    MissingIdentity { facility: &'static str },
120
121    #[error("the artifact carries more than one Harmos {facility} declaration")]
122    DuplicateIdentity { facility: &'static str },
123
124    #[error("a Harmos metadata fragment exceeds its {FRAGMENT_LIMIT}-byte bound")]
125    OversizedFragment,
126
127    #[error("a Harmos metadata fragment is invalid: {source}")]
128    Malformed {
129        #[source]
130        source: serde_json::Error,
131    },
132
133    #[error("a Harmos metadata fragment declares the unknown kind `{kind}`")]
134    UnknownKind { kind: String },
135
136    #[error("`{id}` is not a portable Harmos artifact id")]
137    InvalidId { id: String },
138
139    #[error("the Harmos plugin version is invalid: {source}")]
140    Version {
141        #[source]
142        source: semver::Error,
143    },
144}
145
146#[derive(Deserialize)]
147#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
148enum Fragment {
149    Plugin(contracts::Identity),
150    Contracts(Box<contracts::NamedManifest>),
151    Contributions(Box<contributions::Manifest>),
152    Listeners(Box<listeners::NamedManifest>),
153    Observers(Box<observers::Manifest>),
154    Restart { facility: String, owner: String, policy: RestartPolicy },
155    Publications(publications::NamedManifest),
156    Schema { name: String, fields: Vec<crate::schema::declaration::Field<String>> },
157}
158
159/* === Traits === */
160
161/* === Implementations === */
162
163impl Facility {
164    pub const fn as_str(self) -> &'static str {
165        match self {
166            Self::Sidecar => "sidecar",
167            Self::Guest => "guest",
168        }
169    }
170}
171
172impl Declaration {
173    pub fn inspect(artifact: &[u8], facility: Facility) -> Result<Self, DeclarationError> {
174        let name = facility.as_str();
175        let mut identity = None;
176        let mut schemas = Vec::new();
177        let mut contract_manifests = Vec::new();
178        let mut contribution_manifests = Vec::new();
179        let mut listener_manifests = Vec::new();
180        let mut observer_manifests = Vec::new();
181        let mut publications = Vec::new();
182        let mut restart_policies = Vec::new();
183        let mut settings = crate::config::SchemaSettings::default();
184        for fragment in crate::artifacts::fragments_with_settings::<Fragment>(artifact, &OWNED, &mut settings)? {
185            match fragment {
186                Fragment::Plugin(common) => {
187                    if common.facility != name {
188                        continue;
189                    }
190                    if identity.is_some() {
191                        return Err(DeclarationError::DuplicateIdentity { facility: name });
192                    }
193                    if !artifact_subset(&common.id) {
194                        return Err(DeclarationError::InvalidId { id: common.id });
195                    }
196                    let version = Version::parse(&common.version).map_err(|source| DeclarationError::Version { source })?;
197                    identity = Some((common.id, version, common.description, common.config));
198                }
199                Fragment::Contracts(manifest) => contract_manifests.push(*manifest),
200                Fragment::Contributions(manifest) => contribution_manifests.push(*manifest),
201                Fragment::Listeners(manifest) => listener_manifests.push(*manifest),
202                Fragment::Observers(manifest) => observer_manifests.push(*manifest),
203                Fragment::Restart { facility, owner, policy } => restart_policies.push((facility, owner, policy)),
204                Fragment::Publications(manifest) => publications.push(manifest),
205                Fragment::Schema { name, fields } => schemas.push(crate::schema::declaration::Named { name, fields }),
206            }
207        }
208        let Some((id, version, description, config)) = identity else {
209            return Err(DeclarationError::MissingIdentity { facility: name });
210        };
211
212        // Resolve The Complete Inert Dictionary Before Selecting Owned Sections.
213        let contract_manifests = contract_manifests
214            .into_iter()
215            .map(|manifest| manifest.resolve(&schemas, &mut settings))
216            .collect::<Result<_, contracts::Error>>()?;
217        let listener_manifests = listener_manifests
218            .into_iter()
219            .map(|manifest| manifest.resolve(&schemas, &mut settings))
220            .collect::<Result<_, listeners::Error>>()?;
221        let publications = publications
222            .into_iter()
223            .map(|manifest| manifest.resolve(&schemas, &mut settings))
224            .collect::<Result<_, publications::Error>>()?;
225        let contracts = contracts::select(contract_manifests, name, &id)?;
226        let sections = contributions::select(contribution_manifests, name, &id)?;
227        let listeners = listeners::select(listener_manifests, name, &id, &contracts.provides)?.unwrap_or_default();
228        let operations = contracts.provides.iter().map(|contract| contract.operation().clone()).collect();
229        let streams = publications::select(publications, name, &id)?;
230
231        // Select The CarrierSection Section
232        let (observers, carrier) = match facility {
233            Facility::Sidecar => {
234                let restart = restart(restart_policies, &id)?;
235                let selected = observers::select_manifest(observer_manifests, name, &id, &streams)?;
236                let carrier = CarrierSection::Sidecar { restart };
237                (selected.observers, carrier)
238            }
239            Facility::Guest => {
240                if observer_manifests.iter().any(|manifest| manifest.facility == name && manifest.owner == id) {
241                    let selected = observers::select_manifest(observer_manifests, name, &id, &streams)?;
242                    if !selected.observers.is_empty() {
243                        return Err(DeclarationError::UnsupportedObservers);
244                    }
245                }
246                (Vec::new(), CarrierSection::Guest)
247            }
248        };
249        Ok(Self { id, version, description, config, operations, contracts, listeners, streams, observers, sections, carrier })
250    }
251
252    pub fn section(&self, namespace: &str) -> Option<&Section> {
253        self.sections.iter().find(|section| section.namespace == namespace)
254    }
255
256    pub fn restart(&self) -> Option<RestartPolicy> {
257        match &self.carrier {
258            CarrierSection::Sidecar { restart, .. } => *restart,
259            CarrierSection::Guest => None,
260        }
261    }
262
263    // The built-in `harmos.observe` contract, served by a sidecar that
264    // declares observers.
265    pub fn observe_contract(&self) -> Option<Contract> {
266        if self.observers.is_empty() {
267            return None;
268        }
269        Some(crate::observers::contract().expect("the built-in observe control is valid"))
270    }
271}
272
273impl RestartPolicy {
274    pub fn enabled(self) -> bool {
275        self.credit > 0
276    }
277
278    pub fn bounded(self) -> Self {
279        Self {
280            credit: self.credit.min(10),
281            sustained_health_ms: self.sustained_health_ms.min(60 * 60 * 1_000),
282            backoff_base_ms: self.backoff_base_ms.min(60 * 1_000),
283            backoff_cap_ms: self.backoff_cap_ms.min(5 * 60 * 1_000),
284        }
285    }
286
287    fn validate(self) -> Result<(), DeclarationError> {
288        if self.sustained_health_ms == 0 || self.backoff_base_ms == 0 || self.backoff_cap_ms < self.backoff_base_ms {
289            return Err(DeclarationError::InvalidRestartPolicy);
290        }
291        Ok(())
292    }
293
294    pub fn sustained_health(self) -> Duration {
295        Duration::from_millis(self.bounded().sustained_health_ms)
296    }
297
298    pub fn backoff_base(self) -> Duration {
299        Duration::from_millis(self.bounded().backoff_base_ms)
300    }
301
302    pub fn backoff_cap(self) -> Duration {
303        Duration::from_millis(self.bounded().backoff_cap_ms)
304    }
305}
306
307impl From<ScanError> for DeclarationError {
308    fn from(error: ScanError) -> Self {
309        match error {
310            ScanError::IncompatiblePayload => Self::IncompatiblePayload,
311            ScanError::Budget => Self::Schema(crate::schema::graph::Error::Budget),
312            ScanError::Oversized => Self::OversizedFragment,
313            ScanError::Malformed { source } => Self::Malformed { source },
314            ScanError::UnknownKind { kind } => Self::UnknownKind { kind },
315        }
316    }
317}
318
319/* === Public Functions === */
320
321/* === Private Functions === */
322
323fn restart(policies: Vec<(String, String, RestartPolicy)>, owner: &str) -> Result<Option<RestartPolicy>, DeclarationError> {
324    let mut selected = policies.into_iter().filter(|(facility, candidate, _)| facility == "sidecar" && candidate == owner);
325    let Some((_, _, policy)) = selected.next() else {
326        return Ok(None);
327    };
328    if selected.next().is_some() {
329        return Err(DeclarationError::InvalidRestartPolicy);
330    }
331    policy.validate()?;
332    Ok(Some(policy))
333}
334
335/* === Tests === */
harmos_domain/inventory/declaration.rs

harmos_domain/inventory/
declaration.rs
