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, OperationDefinition};
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<OperationDefinition>,
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(transparent)]
92    Schema(#[from] crate::schema::graph::Error),
93
94    #[error(transparent)]
95    PublicationInventory(#[from] publications::Error),
96
97    #[error(transparent)]
98    ContractInventory(#[from] contracts::Error),
99
100    #[error(transparent)]
101    ContributionInventory(#[from] contributions::Error),
102
103    #[error(transparent)]
104    ListenerInventory(#[from] listeners::Error),
105
106    #[error(transparent)]
107    ObserverInventory(#[from] observers::Error),
108
109    #[error("guest observer callbacks are unsupported")]
110    UnsupportedObservers,
111
112    #[error("the Harmos sidecar restart policy is invalid")]
113    InvalidRestartPolicy,
114
115    #[error("the artifact carries no Harmos {facility} declaration")]
116    MissingIdentity { facility: &'static str },
117
118    #[error("the artifact carries more than one Harmos {facility} declaration")]
119    DuplicateIdentity { facility: &'static str },
120
121    #[error("a Harmos metadata fragment exceeds its {FRAGMENT_LIMIT}-byte bound")]
122    OversizedFragment,
123
124    #[error("a Harmos metadata fragment is invalid: {source}")]
125    Malformed {
126        #[source]
127        source: serde_json::Error,
128    },
129
130    #[error("a Harmos metadata fragment declares the unknown kind `{kind}`")]
131    UnknownKind { kind: String },
132
133    #[error("`{id}` is not a portable Harmos artifact id")]
134    InvalidId { id: String },
135
136    #[error("the Harmos plugin version is invalid: {source}")]
137    Version {
138        #[source]
139        source: semver::Error,
140    },
141}
142
143#[derive(Deserialize)]
144#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
145enum Fragment {
146    Plugin(contracts::Identity),
147    Contracts(Box<contracts::NamedManifest>),
148    Contributions(Box<contributions::Manifest>),
149    Listeners(Box<listeners::NamedManifest>),
150    Observers(Box<observers::Manifest>),
151    Restart { facility: String, owner: String, policy: RestartPolicy },
152    Publications(publications::NamedManifest),
153    Schema { name: String, fields: Vec<crate::schema::declaration::Field<String>> },
154}
155
156/* === Traits === */
157
158/* === Implementations === */
159
160impl Facility {
161    pub const fn as_str(self) -> &'static str {
162        match self {
163            Self::Sidecar => "sidecar",
164            Self::Guest => "guest",
165        }
166    }
167}
168
169impl Declaration {
170    pub fn inspect(artifact: &[u8], facility: Facility) -> Result<Self, DeclarationError> {
171        let name = facility.as_str();
172        let mut identity = None;
173        let mut schemas = Vec::new();
174        let mut contract_manifests = Vec::new();
175        let mut contribution_manifests = Vec::new();
176        let mut listener_manifests = Vec::new();
177        let mut observer_manifests = Vec::new();
178        let mut publications = Vec::new();
179        let mut restart_policies = Vec::new();
180        let mut settings = crate::config::SchemaSettings::default();
181        for fragment in crate::artifacts::fragments_with_settings::<Fragment>(artifact, &OWNED, &mut settings)? {
182            match fragment {
183                Fragment::Plugin(common) => {
184                    if common.facility != name {
185                        continue;
186                    }
187                    if identity.is_some() {
188                        return Err(DeclarationError::DuplicateIdentity { facility: name });
189                    }
190                    if !artifact_subset(&common.id) {
191                        return Err(DeclarationError::InvalidId { id: common.id });
192                    }
193                    let version = Version::parse(&common.version).map_err(|source| DeclarationError::Version { source })?;
194                    identity = Some((common.id, version, common.description, common.config));
195                }
196                Fragment::Contracts(manifest) => contract_manifests.push(*manifest),
197                Fragment::Contributions(manifest) => contribution_manifests.push(*manifest),
198                Fragment::Listeners(manifest) => listener_manifests.push(*manifest),
199                Fragment::Observers(manifest) => observer_manifests.push(*manifest),
200                Fragment::Restart { facility, owner, policy } => restart_policies.push((facility, owner, policy)),
201                Fragment::Publications(manifest) => publications.push(manifest),
202                Fragment::Schema { name, fields } => schemas.push(crate::schema::declaration::Named { name, fields }),
203            }
204        }
205        let Some((id, version, description, config)) = identity else {
206            return Err(DeclarationError::MissingIdentity { facility: name });
207        };
208
209        // Resolve The Complete Inert Dictionary Before Selecting Owned Sections.
210        let contract_manifests = contract_manifests
211            .into_iter()
212            .map(|manifest| manifest.resolve(&schemas, &mut settings))
213            .collect::<Result<_, contracts::Error>>()?;
214        let listener_manifests = listener_manifests
215            .into_iter()
216            .map(|manifest| manifest.resolve(&schemas, &mut settings))
217            .collect::<Result<_, listeners::Error>>()?;
218        let publications = publications
219            .into_iter()
220            .map(|manifest| manifest.resolve(&schemas, &mut settings))
221            .collect::<Result<_, publications::Error>>()?;
222        let contracts = contracts::select(contract_manifests, name, &id)?;
223        let sections = contributions::select(contribution_manifests, name, &id)?;
224        let listeners = listeners::select(listener_manifests, name, &id, &contracts.provides)?.unwrap_or_default();
225        let operations = contracts.provides.iter().map(|contract| contract.operation().clone()).collect();
226        let streams = publications::select(publications, name, &id)?;
227
228        // Select The CarrierSection Section
229        let (observers, carrier) = match facility {
230            Facility::Sidecar => {
231                let restart = restart(restart_policies, &id)?;
232                let selected = observers::select_manifest(observer_manifests, name, &id, &streams)?;
233                let carrier = CarrierSection::Sidecar { restart };
234                (selected.observers, carrier)
235            }
236            Facility::Guest => {
237                if observer_manifests.iter().any(|manifest| manifest.facility == name && manifest.owner == id) {
238                    let selected = observers::select_manifest(observer_manifests, name, &id, &streams)?;
239                    if !selected.observers.is_empty() {
240                        return Err(DeclarationError::UnsupportedObservers);
241                    }
242                }
243                (Vec::new(), CarrierSection::Guest)
244            }
245        };
246        Ok(Self {
247            id,
248            version,
249            description,
250            config,
251            operations,
252            contracts,
253            listeners,
254            streams,
255            observers,
256            sections,
257            carrier,
258        })
259    }
260
261    pub fn section(&self, namespace: &str) -> Option<&Section> {
262        self.sections.iter().find(|section| section.namespace == namespace)
263    }
264
265    pub fn restart(&self) -> Option<RestartPolicy> {
266        match &self.carrier {
267            CarrierSection::Sidecar { restart, .. } => *restart,
268            CarrierSection::Guest => None,
269        }
270    }
271
272    // The built-in `harmos.observe` contract, served by a sidecar that
273    // declares observers.
274    pub fn observe_contract(&self) -> Option<Contract> {
275        if self.observers.is_empty() {
276            return None;
277        }
278        Some(crate::observers::contract().expect("the built-in observe control is valid"))
279    }
280}
281
282impl RestartPolicy {
283    pub fn enabled(self) -> bool {
284        self.credit > 0
285    }
286
287    pub fn bounded(self) -> Self {
288        Self {
289            credit: self.credit.min(10),
290            sustained_health_ms: self.sustained_health_ms.min(60 * 60 * 1_000),
291            backoff_base_ms: self.backoff_base_ms.min(60 * 1_000),
292            backoff_cap_ms: self.backoff_cap_ms.min(5 * 60 * 1_000),
293        }
294    }
295
296    fn validate(self) -> Result<(), DeclarationError> {
297        if self.sustained_health_ms == 0 || self.backoff_base_ms == 0 || self.backoff_cap_ms < self.backoff_base_ms {
298            return Err(DeclarationError::InvalidRestartPolicy);
299        }
300        Ok(())
301    }
302
303    pub fn sustained_health(self) -> Duration {
304        Duration::from_millis(self.bounded().sustained_health_ms)
305    }
306
307    pub fn backoff_base(self) -> Duration {
308        Duration::from_millis(self.bounded().backoff_base_ms)
309    }
310
311    pub fn backoff_cap(self) -> Duration {
312        Duration::from_millis(self.bounded().backoff_cap_ms)
313    }
314}
315
316impl From<ScanError> for DeclarationError {
317    fn from(error: ScanError) -> Self {
318        match error {
319            ScanError::Budget => Self::Schema(crate::schema::graph::Error::Budget),
320            ScanError::Oversized => Self::OversizedFragment,
321            ScanError::Malformed { source } => Self::Malformed { source },
322            ScanError::UnknownKind { kind } => Self::UnknownKind { kind },
323        }
324    }
325}
326
327/* === Public Functions === */
328
329/* === Private Functions === */
330
331fn restart(policies: Vec<(String, String, RestartPolicy)>, owner: &str) -> Result<Option<RestartPolicy>, DeclarationError> {
332    let mut selected = policies.into_iter().filter(|(facility, candidate, _)| facility == "sidecar" && candidate == owner);
333    let Some((_, _, policy)) = selected.next() else {
334        return Ok(None);
335    };
336    if selected.next().is_some() {
337        return Err(DeclarationError::InvalidRestartPolicy);
338    }
339    policy.validate()?;
340    Ok(Some(policy))
341}
342
343/* === Tests === */
harmos_domain/inventory/declaration.rs

harmos_domain/inventory/
declaration.rs
