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};
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 !crate::identity::segment(&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        let (observers, carrier) = match facility {
229            Facility::Sidecar => {
230                let restart = restart(restart_policies, &id)?;
231                let selected = observers::select_manifest(observer_manifests, name, &id, &streams)?;
232                let carrier = CarrierSection::Sidecar { restart };
233                (selected.observers, carrier)
234            }
235            Facility::Guest => {
236                if observer_manifests.iter().any(|manifest| manifest.facility == name && manifest.owner == id) {
237                    let selected = observers::select_manifest(observer_manifests, name, &id, &streams)?;
238                    if !selected.observers.is_empty() {
239                        return Err(DeclarationError::UnsupportedObservers);
240                    }
241                }
242                (Vec::new(), CarrierSection::Guest)
243            }
244        };
245        Ok(Self {
246            id,
247            version,
248            description,
249            config,
250            operations,
251            contracts,
252            listeners,
253            streams,
254            observers,
255            sections,
256            carrier,
257        })
258    }
259
260    pub fn section(&self, namespace: &str) -> Option<&Section> {
261        self.sections.iter().find(|section| section.namespace == namespace)
262    }
263
264    pub fn restart(&self) -> Option<RestartPolicy> {
265        match &self.carrier {
266            CarrierSection::Sidecar { restart, .. } => *restart,
267            CarrierSection::Guest => None,
268        }
269    }
270
271    // The built-in `harmos.observe` contract, served by a sidecar that
272    // declares observers.
273    pub fn observe_contract(&self) -> Option<Contract> {
274        if self.observers.is_empty() {
275            return None;
276        }
277        Some(crate::observers::contract().expect("the built-in observe control is valid"))
278    }
279}
280
281impl RestartPolicy {
282    pub fn enabled(self) -> bool {
283        self.credit > 0
284    }
285
286    pub fn bounded(self) -> Self {
287        Self {
288            credit: self.credit.min(10),
289            sustained_health_ms: self.sustained_health_ms.min(60 * 60 * 1_000),
290            backoff_base_ms: self.backoff_base_ms.min(60 * 1_000),
291            backoff_cap_ms: self.backoff_cap_ms.min(5 * 60 * 1_000),
292        }
293    }
294
295    fn validate(self) -> Result<(), DeclarationError> {
296        if self.sustained_health_ms == 0 || self.backoff_base_ms == 0 || self.backoff_cap_ms < self.backoff_base_ms {
297            return Err(DeclarationError::InvalidRestartPolicy);
298        }
299        Ok(())
300    }
301
302    pub fn sustained_health(self) -> Duration {
303        Duration::from_millis(self.bounded().sustained_health_ms)
304    }
305
306    pub fn backoff_base(self) -> Duration {
307        Duration::from_millis(self.bounded().backoff_base_ms)
308    }
309
310    pub fn backoff_cap(self) -> Duration {
311        Duration::from_millis(self.bounded().backoff_cap_ms)
312    }
313}
314
315impl From<ScanError> for DeclarationError {
316    fn from(error: ScanError) -> Self {
317        match error {
318            ScanError::Budget => Self::Schema(crate::schema::graph::Error::Budget),
319            ScanError::Oversized => Self::OversizedFragment,
320            ScanError::Malformed { source } => Self::Malformed { source },
321            ScanError::UnknownKind { kind } => Self::UnknownKind { kind },
322        }
323    }
324}
325
326/* === Public Functions === */
327
328/* === Private Functions === */
329
330fn restart(policies: Vec<(String, String, RestartPolicy)>, owner: &str) -> Result<Option<RestartPolicy>, DeclarationError> {
331    let mut selected = policies.into_iter().filter(|(facility, candidate, _)| facility == "sidecar" && candidate == owner);
332    let Some((_, _, policy)) = selected.next() else {
333        return Ok(None);
334    };
335    if selected.next().is_some() {
336        return Err(DeclarationError::InvalidRestartPolicy);
337    }
338    policy.validate()?;
339    Ok(Some(policy))
340}
341
342/* === Tests === */
harmos_domain/inventory/declaration.rs

harmos_domain/inventory/
declaration.rs
