harmos_domain/operations/common/
contract.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::collections::BTreeSet;
18use std::sync::OnceLock;
19// Third Party Imports
20use serde::{Deserialize, Serialize};
21// Local Imports
22use super::{Definition, Entry, OperationDeclaration};
23use crate::catalog::{Catalog, Member, Projection, Visitor};
24use crate::config::SchemaSettings;
25use crate::schema::declaration::Definition as SchemaDefinition;
26use crate::schema::graph::{Kind, Schema, View};
27use crate::schema::{self, Described};
28
29/* === Constants === */
30
31static INITIALIZE_CONTRACT: OnceLock<Contract> = OnceLock::new();
32static SHUTDOWN_CONTRACT: OnceLock<Contract> = OnceLock::new();
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38/* === Structs === */
39
40#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(try_from = "DecodedContract")]
42pub struct Contract {
43    operation: Definition,
44    version: u32,
45    codec: Codec,
46    schema: Schema,
47
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    grpc: Option<super::Grpc>,
50}
51
52#[derive(Deserialize)]
53#[serde(deny_unknown_fields)]
54struct DecodedContract {
55    #[serde(with = "super::DecodedDefinition")]
56    operation: Definition,
57
58    version: u32,
59    codec: Codec,
60    schema: Schema,
61
62    #[serde(default)]
63    grpc: Option<super::Grpc>,
64}
65
66pub struct WireDeclaration {
67    pub operation: Entry,
68    pub version: u32,
69    pub request: SchemaDefinition,
70    pub response: SchemaDefinition,
71    pub input: Option<SchemaDefinition>,
72    pub output: Option<SchemaDefinition>,
73    pub grpc: Option<super::Grpc>,
74}
75
76pub struct Contracts;
77
78// Wire Contracts Projected From Resource-Bound Operation Handlers.
79
80/* === Enums === */
81
82#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
83#[serde(rename_all = "snake_case")]
84pub enum Codec {
85    Postcard,
86}
87
88#[derive(Debug, thiserror::Error)]
89pub enum ContractError {
90    #[error("invalid or reserved gRPC binding name")]
91    Grpc,
92
93    #[error("invalid wire operation metadata: {0}")]
94    Operation(#[from] crate::Error),
95
96    #[error("wire operation `{id}` has a zero semantic version")]
97    Version { id: String },
98
99    #[error("invalid {direction} schema for wire operation `{id}`: {source}")]
100    Schema {
101        id: String,
102        direction: &'static str,
103
104        #[source]
105        source: schema::Error,
106    },
107}
108
109/* === Traits === */
110
111/* === Implementations === */
112
113impl WireDeclaration {
114    pub const fn of<O: OperationDeclaration>() -> Self {
115        Self {
116            operation: Entry::of::<O>(),
117            version: O::VERSION,
118            request: O::DEFINITION,
119            response: <O::Response as Described>::DEFINITION,
120            input: direction::<O::Input>(),
121            output: direction::<O::Output>(),
122            grpc: O::GRPC,
123        }
124    }
125
126    pub fn contract(&self) -> Result<Contract, ContractError> {
127        let operation = self.operation.definition()?;
128        if self.version == 0 {
129            return Err(ContractError::Version { id: operation.id.to_string() });
130        }
131        let mut roots = vec![("request", self.request), ("response", self.response)];
132        if let Some(input) = self.input {
133            roots.push(("input", input));
134        }
135        if let Some(output) = self.output {
136            roots.push(("output", output));
137        }
138        let schema = Schema::from_declarations(&roots, &SchemaSettings::default()).map_err(|source| ContractError::Schema {
139            id: operation.id.to_string(),
140            direction: "contract",
141            source: source.into(),
142        })?;
143        Contract::new(operation, self.version, Codec::Postcard, schema)?.with_grpc(self.grpc.clone())
144    }
145}
146
147impl Contract {
148    pub fn new(operation: Definition, version: u32, codec: Codec, schema: Schema) -> Result<Self, ContractError> {
149        operation.validate()?;
150        Self::canonical(operation, version, codec, schema)
151    }
152
153    fn canonical(operation: Definition, version: u32, codec: Codec, schema: Schema) -> Result<Self, ContractError> {
154        if version == 0 {
155            return Err(ContractError::Version { id: operation.id.to_string() });
156        }
157        let invalid = |source: schema::graph::Error| ContractError::Schema {
158            id: operation.id.to_string(),
159            direction: "contract",
160            source: source.into(),
161        };
162        schema.prepare(&SchemaSettings::default()).map_err(invalid)?;
163        let directions = schema.root().map_err(invalid)?;
164        let fields = directions.fields();
165        let valid = (2..=4).contains(&fields.len())
166            && fields[0].name == "request"
167            && fields[1].name == "response"
168            && fields.iter().enumerate().all(|(index, field)| {
169                field.tag == (index + 1) as u32
170                    && field.cardinality == schema::Cardinality::Required
171                    && matches!(field.kind, Kind::Message { .. })
172            })
173            && fields.get(2).is_none_or(|field| field.name == "input" || field.name == "output")
174            && fields.get(3).is_none_or(|field| fields[2].name == "input" && field.name == "output");
175        if !valid {
176            return Err(invalid(schema::graph::Error::Field {
177                field: "contract directions".into(),
178                reason: "expected request, response, and optional input/output roots",
179            }));
180        }
181        Ok(Self { operation, version, codec, schema, grpc: None })
182    }
183
184    pub fn grpc(&self) -> Option<&super::Grpc> {
185        self.grpc.as_ref()
186    }
187
188    pub fn with_grpc(mut self, binding: Option<super::Grpc>) -> Result<Self, ContractError> {
189        if let Some(binding) = &binding {
190            binding.validate()?;
191        }
192        self.grpc = binding;
193        Ok(self)
194    }
195
196    // Only The Closed SDK Action Can Construct Reserved Metadata
197    pub(crate) fn lifecycle(action: crate::operations::lifecycle::Action, policy: super::OperationPolicy) -> Result<Self, ContractError> {
198        use crate::operations::lifecycle::{self, Action, Initialize, Shutdown};
199        policy.validate().map_err(crate::Error::from)?;
200        let cache = match action {
201            Action::Initialize => &INITIALIZE_CONTRACT,
202            Action::Shutdown => &SHUTDOWN_CONTRACT,
203        };
204        let template = if let Some(contract) = cache.get() {
205            contract
206        } else {
207            let contract = match action {
208                Action::Initialize => Self::reserved::<Initialize>(lifecycle::POLICY)?,
209                Action::Shutdown => Self::reserved::<Shutdown>(lifecycle::POLICY)?,
210            };
211            cache.get_or_init(|| contract)
212        };
213        let mut contract = template.clone();
214        contract.operation.policy = policy;
215        Ok(contract)
216    }
217
218    // The Contract Of A Built-In Operation Under The Reserved `harmos.` Namespace,
219    // Which Ordinary Declarations Cannot Claim.
220    pub(crate) fn reserved<O: OperationDeclaration>(policy: super::OperationPolicy) -> Result<Self, ContractError> {
221        policy.validate().map_err(crate::Error::from)?;
222        let operation = Definition {
223            id: crate::Identifier::new(O::ID).map_err(crate::Error::from)?,
224            description: O::DESCRIPTION.to_owned(),
225            policy,
226            visibility: O::VISIBILITY,
227            replay: O::REPLAY,
228        };
229        let schema = Schema::from_declarations(
230            &[("request", O::DEFINITION), ("response", <O::Response as Described>::DEFINITION)],
231            &SchemaSettings::default(),
232        )
233        .map_err(|source| ContractError::Schema { id: operation.id.to_string(), direction: "contract", source: source.into() })?;
234        Self::canonical(operation, O::VERSION, Codec::Postcard, schema)
235    }
236
237    pub fn operation(&self) -> &Definition {
238        &self.operation
239    }
240
241    pub fn version(&self) -> u32 {
242        self.version
243    }
244
245    pub fn codec(&self) -> Codec {
246        self.codec
247    }
248
249    pub fn schema(&self) -> &Schema {
250        &self.schema
251    }
252
253    pub fn request(&self) -> View<'_> {
254        self.direction("request").expect("validated request root")
255    }
256
257    pub fn response(&self) -> View<'_> {
258        self.direction("response").expect("validated response root")
259    }
260
261    pub fn input(&self) -> Option<View<'_>> {
262        self.direction("input")
263    }
264
265    pub fn output(&self) -> Option<View<'_>> {
266        self.direction("output")
267    }
268
269    pub fn compatible(&self, requirement: &Self) -> bool {
270        self.operation.id == requirement.operation.id
271            && self.version == requirement.version
272            && self.codec == requirement.codec
273            && ["request", "response", "input", "output"].into_iter().all(|name| {
274                match (self.direction(name), requirement.direction(name)) {
275                    (None, None) => true,
276                    (Some(left), Some(right)) => self
277                        .schema
278                        .equivalent(left.index(), &requirement.schema, right.index(), &SchemaSettings::default())
279                        .unwrap_or(false),
280                    _ => false,
281                }
282            })
283    }
284
285    fn direction(&self, name: &str) -> Option<View<'_>> {
286        let field = self.schema.definitions[self.schema.root].iter().find(|field| field.name == name)?;
287        let Kind::Message { definition } = field.kind else {
288            unreachable!("validated contract direction");
289        };
290        Some(self.schema.view(definition).expect("validated direction reference"))
291    }
292}
293
294impl TryFrom<DecodedContract> for Contract {
295    type Error = ContractError;
296
297    fn try_from(decoded: DecodedContract) -> Result<Self, Self::Error> {
298        if !crate::reserved(decoded.operation.id.as_str()) {
299            return Self::new(decoded.operation, decoded.version, decoded.codec, decoded.schema)?.with_grpc(decoded.grpc);
300        }
301
302        // Read Only Exact SDK Metadata; Ordinary Declaration Admission Still Rejects These Names.
303        use crate::operations::lifecycle::{Action, Descriptor, Initialize, Shutdown};
304        let refused = || crate::Error::ReservedOperation { id: decoded.operation.id.to_string() }.into();
305        let expected = match decoded.operation.id.as_str() {
306            crate::observers::Control::ID => crate::observers::contract()?,
307            Initialize::ID => Descriptor::new(Action::Initialize, decoded.operation.policy).map_err(|_| refused())?.contract().clone(),
308            Shutdown::ID => Descriptor::new(Action::Shutdown, decoded.operation.policy).map_err(|_| refused())?.contract().clone(),
309            _ => return Err(refused()),
310        };
311        let identity_matches =
312            decoded.operation == expected.operation && decoded.version == expected.version && decoded.codec == expected.codec;
313        let schema_matches = decoded.schema == expected.schema && decoded.grpc == expected.grpc;
314        if !identity_matches || !schema_matches {
315            return Err(refused());
316        }
317        Ok(expected)
318    }
319}
320
321impl Visitor for Contracts {
322    type Entry = WireDeclaration;
323}
324
325impl<M: super::Declares> Member<M> for Contracts {
326    const ENTRY: WireDeclaration = WireDeclaration::of::<M::Declaration>();
327}
328
329impl Contracts {
330    pub fn of<C: Catalog<Self>>() -> Result<Vec<Contract>, ContractError> {
331        Self::from_projection(&C::ENTRIES)
332    }
333
334    pub fn from_entries(entries: &[WireDeclaration]) -> Result<Vec<Contract>, ContractError> {
335        let mut ids = BTreeSet::new();
336        let mut contracts = Vec::with_capacity(entries.len());
337        for declaration in entries {
338            append(declaration, &mut ids, &mut contracts)?;
339        }
340        Ok(contracts)
341    }
342
343    pub fn from_tree(entries: &[crate::catalog::Node<WireDeclaration>]) -> Result<Vec<Contract>, ContractError> {
344        let mut ids = BTreeSet::new();
345        let mut contracts = Vec::new();
346        crate::catalog::try_for_each(entries, |declaration| append(declaration, &mut ids, &mut contracts))?;
347        Ok(contracts)
348    }
349
350    pub fn from_projection<P: Projection<WireDeclaration>>(entries: &P) -> Result<Vec<Contract>, ContractError> {
351        let mut ids = BTreeSet::new();
352        let mut contracts = Vec::with_capacity(P::LEN);
353        entries.try_for_each(|declaration| append(declaration, &mut ids, &mut contracts))?;
354        Ok(contracts)
355    }
356}
357
358/* === Public Functions === */
359
360/* === Private Functions === */
361
362const fn direction<D: super::Direction>() -> Option<SchemaDefinition> {
363    if D::STREAMED { Some(<D::Item as Described>::DEFINITION) } else { None }
364}
365
366fn append(
367    declaration: &WireDeclaration, ids: &mut BTreeSet<crate::Identifier>, contracts: &mut Vec<Contract>,
368) -> Result<(), ContractError> {
369    let contract = declaration.contract()?;
370    if !ids.insert(contract.operation.id.clone()) {
371        return Err(crate::Error::DuplicateOperation { id: contract.operation.id.to_string() }.into());
372    }
373    contracts.push(contract);
374    Ok(())
375}
376
377/* === Tests === */
378
379#[cfg(test)]
380mod tests {
381    use super::*;
382    use crate::operations::lifecycle::{self, Action, Descriptor};
383    use crate::operations::{Backoff, Jitter, RestartPolicy, RetryPolicy};
384    use serde_json::json;
385
386    fn sdk_contracts() -> [Contract; 3] {
387        let policy = super::super::OperationPolicy {
388            timeout: std::time::Duration::from_secs(17),
389            attempt_timeout: std::time::Duration::from_secs(11),
390            ..lifecycle::POLICY
391        };
392        [
393            crate::observers::contract().unwrap(),
394            Descriptor::new(Action::Initialize, policy).unwrap().contract().clone(),
395            Descriptor::new(Action::Shutdown, policy).unwrap().contract().clone(),
396        ]
397    }
398
399    #[test]
400    fn sdk_metadata_roundtrip() {
401        for original in sdk_contracts() {
402            let encoded = serde_json::to_vec(&original).unwrap();
403            let decoded: Contract = serde_json::from_slice(&encoded).unwrap();
404            assert_eq!(decoded, original);
405        }
406    }
407
408    #[test]
409    fn sdk_metadata_forgery() {
410        let delay = std::time::Duration::from_secs(1);
411        let backoff = Backoff::Fixed { delay, jitter: Jitter::None };
412        for original in sdk_contracts() {
413            let encoded = serde_json::to_value(&original).unwrap();
414            for (pointer, replacement) in [
415                ("/operation/id", json!("harmos.forged")),
416                ("/operation/description", json!("forged SDK contract")),
417                ("/operation/visibility", json!("shared")),
418                ("/operation/policy/attempt_timeout/secs", json!(0)),
419                ("/operation/policy/retries", json!(RetryPolicy::OnTransient { max_retries: 1, backoff })),
420                ("/operation/policy/restart", json!(RestartPolicy::OnTransient { credit: 1, sustained_health: delay, backoff })),
421                ("/version", json!(2)),
422                ("/schema", json!({"root": 0, "definitions": [[]]})),
423            ] {
424                let mut forged = encoded.clone();
425                *forged.pointer_mut(pointer).unwrap() = replacement;
426                assert!(serde_json::from_value::<Contract>(forged).is_err(), "{pointer}");
427            }
428            let mut exposed = encoded;
429            exposed["grpc"] = json!({"package": "forged.v1", "service": "Exposed", "method": "Invoke"});
430            assert!(serde_json::from_value::<Contract>(exposed).is_err());
431        }
432    }
433
434    #[test]
435    fn sdk_lifecycle_refuses_invalid() {
436        let delay = std::time::Duration::from_secs(1);
437        let backoff = Backoff::Fixed { delay, jitter: Jitter::None };
438        let descriptors = [
439            Descriptor::new(Action::Initialize, lifecycle::POLICY).unwrap(),
440            Descriptor::new(Action::Shutdown, lifecycle::POLICY).unwrap(),
441        ];
442        for descriptor in &descriptors {
443            assert_eq!(Contract::from(descriptor.clone()), *descriptor.contract());
444            let encoded = serde_json::to_value(descriptor).unwrap();
445            assert_eq!(serde_json::from_value::<Descriptor>(encoded.clone()).unwrap(), *descriptor);
446            for (pointer, replacement) in [
447                ("/version", json!(2)),
448                ("/policy/attempt_timeout/secs", json!(0)),
449                ("/policy/retries", json!(RetryPolicy::OnTransient { max_retries: 1, backoff })),
450                ("/policy/restart", json!(RestartPolicy::OnTransient { credit: 1, sustained_health: delay, backoff })),
451            ] {
452                let mut forged = encoded.clone();
453                *forged.pointer_mut(pointer).unwrap() = replacement;
454                assert!(serde_json::from_value::<Descriptor>(forged).is_err(), "{pointer}");
455            }
456        }
457        let mut manifest = crate::inventory::contracts::Manifest {
458            facility: "guest".into(),
459            owner: "fixture".into(),
460            provides: Vec::new(),
461            requires: Vec::new(),
462            lifecycle: descriptors,
463        };
464        crate::inventory::contracts::validate(&manifest).unwrap();
465        manifest.lifecycle.swap(0, 1);
466        assert!(crate::inventory::contracts::validate(&manifest).is_err());
467    }
468
469    #[test]
470    fn reserved_operations_refused() {
471        use crate::inventory::contracts::{Manifest, validate};
472        for contract in sdk_contracts() {
473            assert!(contract.operation().validate().is_err());
474            let definition = serde_json::to_value(contract.operation()).unwrap();
475            assert!(serde_json::from_value::<Definition>(definition).is_err());
476
477            let mut manifest = Manifest {
478                facility: "sidecar".into(),
479                owner: "fixture".into(),
480                provides: Vec::new(),
481                requires: Vec::new(),
482                lifecycle: [
483                    Descriptor::new(Action::Initialize, lifecycle::POLICY).unwrap(),
484                    Descriptor::new(Action::Shutdown, lifecycle::POLICY).unwrap(),
485                ],
486            };
487            validate(&manifest).unwrap();
488            manifest.provides.push(contract);
489            assert!(validate(&manifest).is_err());
490            let encoded = serde_json::to_value(&manifest).unwrap();
491            assert!(serde_json::from_value::<Manifest>(encoded).is_err());
492        }
493    }
494}
harmos_domain/operations/common/contract.rs

harmos_domain/operations/common/
contract.rs
