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

harmos_domain/operations/common/
contract.rs
