harmos_domain/operations/
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::{Direction, Entry, OperationDeclaration, OperationDefinition};
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: OperationDefinition,
44    fingerprint: crate::Fingerprint,
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::DecodedOperationDefinition")]
56    operation: OperationDefinition,
57
58    fingerprint: crate::Fingerprint,
59    codec: Codec,
60    schema: Schema,
61
62    #[serde(default)]
63    grpc: Option<super::Grpc>,
64}
65
66#[derive(Clone)]
67pub struct WireDeclaration {
68    pub operation: Entry,
69    pub codec: &'static str,
70    pub response_codec: &'static str,
71    pub input_codec: Option<&'static str>,
72    pub output_codec: Option<&'static str>,
73    pub request: SchemaDefinition,
74    pub response: SchemaDefinition,
75    pub input: Option<SchemaDefinition>,
76    pub output: Option<SchemaDefinition>,
77    pub grpc: Option<super::Grpc>,
78}
79
80// Wire Contracts Projected From Resource-Bound Operation Handlers.
81pub struct Contracts;
82
83/* === Enums === */
84
85#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
86#[serde(rename_all = "snake_case")]
87pub enum Codec {
88    Postcard,
89}
90
91#[derive(Debug, thiserror::Error)]
92pub enum ContractError {
93    #[error("invalid or reserved gRPC binding name")]
94    Grpc,
95
96    #[error("invalid wire operation metadata: {0}")]
97    Operation(#[from] crate::Error),
98
99    #[error("wire operation `{id}` has an invalid contract fingerprint")]
100    Fingerprint { id: String },
101
102    #[error("wire operation uses unsupported codec `{codec}`")]
103    Codec { codec: String },
104
105    #[error("invalid {direction} schema for wire operation `{id}`: {source}")]
106    Schema {
107        id: String,
108        direction: &'static str,
109
110        #[source]
111        source: schema::Error,
112    },
113}
114
115/* === Traits === */
116
117/* === Implementations === */
118
119impl Codec {
120    pub const fn id(self) -> &'static str {
121        match self {
122            Self::Postcard => "harmos.postcard.v1",
123        }
124    }
125}
126
127impl WireDeclaration {
128    pub const fn copied(&self) -> Self {
129        let grpc = match &self.grpc {
130            None => None,
131            Some(grpc) => {
132                let (std::borrow::Cow::Borrowed(package), std::borrow::Cow::Borrowed(service), std::borrow::Cow::Borrowed(method)) =
133                    (&grpc.package, &grpc.service, &grpc.method)
134                else {
135                    panic!("static gRPC declarations must borrow names");
136                };
137                Some(super::Grpc::new(package, service, method))
138            }
139        };
140        Self {
141            operation: self.operation,
142            codec: self.codec,
143            response_codec: self.response_codec,
144            input_codec: self.input_codec,
145            output_codec: self.output_codec,
146            request: self.request,
147            response: self.response,
148            input: self.input,
149            output: self.output,
150            grpc,
151        }
152    }
153
154    pub const fn of<O: OperationDeclaration>() -> Self {
155        Self {
156            operation: Entry::of::<O>(),
157            codec: <O::Request as crate::Message>::CODEC,
158            response_codec: <O::Response as crate::Message>::CODEC,
159            input_codec: if O::Input::STREAMED { Some(<<O::Input as super::Direction>::Item as crate::Message>::CODEC) } else { None },
160            output_codec: if O::Output::STREAMED { Some(<<O::Output as super::Direction>::Item as crate::Message>::CODEC) } else { None },
161            request: <O::Request as Described>::DEFINITION,
162            response: <O::Response as Described>::DEFINITION,
163            input: direction::<O::Input>(),
164            output: direction::<O::Output>(),
165            grpc: O::GRPC,
166        }
167    }
168
169    pub fn contract(&self) -> Result<Contract, ContractError> {
170        self.contract_with_settings(&SchemaSettings::default())
171    }
172
173    pub fn contract_with_settings(&self, settings: &SchemaSettings) -> Result<Contract, ContractError> {
174        let operation = self.operation.definition()?;
175        for codec in [Some(self.codec), Some(self.response_codec), self.input_codec, self.output_codec].into_iter().flatten() {
176            if codec != "harmos.postcard.v1" {
177                return Err(ContractError::Codec { codec: codec.to_owned() });
178            }
179        }
180        let mut roots = vec![("request", self.request), ("response", self.response)];
181        if let Some(input) = self.input {
182            roots.push(("input", input));
183        }
184        if let Some(output) = self.output {
185            roots.push(("output", output));
186        }
187        let schema = Schema::from_declarations(&roots, settings).map_err(|source| ContractError::Schema {
188            id: operation.id.to_string(),
189            direction: "contract",
190            source: source.into(),
191        })?;
192        Contract::new_with_settings(operation.clone(), operation.fingerprint, Codec::Postcard, schema, settings)?
193            .with_grpc(self.grpc.clone())
194    }
195}
196
197impl Contract {
198    pub fn new(
199        operation: OperationDefinition, fingerprint: crate::Fingerprint, codec: Codec, schema: Schema,
200    ) -> Result<Self, ContractError> {
201        Self::new_with_settings(operation, fingerprint, codec, schema, &SchemaSettings::default())
202    }
203
204    pub fn new_with_settings(
205        operation: OperationDefinition, fingerprint: crate::Fingerprint, codec: Codec, schema: Schema, settings: &SchemaSettings,
206    ) -> Result<Self, ContractError> {
207        operation.validate()?;
208        Self::canonical(operation, fingerprint, codec, schema, settings)
209    }
210
211    fn canonical(
212        operation: OperationDefinition, fingerprint: crate::Fingerprint, codec: Codec, schema: Schema, settings: &SchemaSettings,
213    ) -> Result<Self, ContractError> {
214        let invalid = |source: schema::graph::Error| ContractError::Schema {
215            id: operation.id.to_string(),
216            direction: "contract",
217            source: source.into(),
218        };
219        schema.prepare(settings).map_err(invalid)?;
220        let directions = schema.root().map_err(invalid)?;
221        let fields = directions.fields();
222        let valid = (2..=4).contains(&fields.len())
223            && fields[0].name == "request"
224            && fields[1].name == "response"
225            && fields.iter().enumerate().all(|(index, field)| {
226                field.tag == (index + 1) as u32
227                    && field.cardinality == schema::Cardinality::Required
228                    && matches!(field.kind, Kind::Message { .. })
229            })
230            && fields.get(2).is_none_or(|field| field.name == "input" || field.name == "output")
231            && fields.get(3).is_none_or(|field| fields[2].name == "input" && field.name == "output");
232        if !valid {
233            return Err(invalid(schema::graph::Error::Field {
234                field: "contract directions".into(),
235                reason: "expected request, response, and optional input/output roots",
236            }));
237        }
238        let contract = Self {
239            operation,
240            fingerprint,
241            codec,
242            schema,
243            grpc: None,
244        };
245        let computed = contract.computed_fingerprint()?;
246        if fingerprint != computed || contract.operation.fingerprint != computed {
247            return Err(ContractError::Fingerprint { id: contract.operation.id.to_string() });
248        }
249        Ok(contract)
250    }
251
252    pub fn grpc(&self) -> Option<&super::Grpc> {
253        self.grpc.as_ref()
254    }
255
256    pub fn with_grpc(mut self, binding: Option<super::Grpc>) -> Result<Self, ContractError> {
257        if let Some(binding) = &binding {
258            binding.validate()?;
259        }
260        self.grpc = binding;
261        Ok(self)
262    }
263
264    // Only The Closed SDK Action Can Construct Reserved Metadata
265    pub(crate) fn lifecycle(
266        action: crate::model::operations::lifecycle::Action, policy: super::OperationPolicy,
267    ) -> Result<Self, ContractError> {
268        use crate::model::operations::lifecycle::{self, Action};
269        policy.validate().map_err(crate::Error::from)?;
270        let cache = match action {
271            Action::Initialize => &INITIALIZE_CONTRACT,
272            Action::Shutdown => &SHUTDOWN_CONTRACT,
273        };
274        let template = if let Some(contract) = cache.get() {
275            contract
276        } else {
277            let request = action.request();
278            let request_schema = Schema::from_declaration(request, &SchemaSettings::default()).map_err(|source| ContractError::Schema {
279                id: action.id().into(),
280                direction: "request",
281                source: source.into(),
282            })?;
283            let request_root = request_schema.root().map_err(|source| ContractError::Schema {
284                id: action.id().into(),
285                direction: "request",
286                source: source.into(),
287            })?;
288            let request_fingerprint =
289                crate::schema::identity::fingerprint(request_root, Codec::Postcard.id()).map_err(|source| ContractError::Schema {
290                    id: action.id().into(),
291                    direction: "request",
292                    source: source.into(),
293                })?;
294            let fingerprint =
295                crate::schema::identity::operation(action.id(), request_fingerprint, <() as crate::Portable>::fingerprint(), None, None);
296            let operation = OperationDefinition {
297                id: crate::Identifier::new(action.id()).map_err(crate::Error::from)?,
298                description: String::new(),
299                policy: lifecycle::POLICY,
300                visibility: super::Visibility::Private,
301                replay: super::ReplaySafety::Unknown,
302                fingerprint,
303            };
304            let schema =
305                Schema::from_declarations(&[("request", request), ("response", <() as Described>::DEFINITION)], &SchemaSettings::default())
306                    .map_err(|source| ContractError::Schema {
307                        id: action.id().into(),
308                        direction: "contract",
309                        source: source.into(),
310                    })?;
311            let contract = Self::canonical(operation, fingerprint, Codec::Postcard, schema, &SchemaSettings::default())?;
312            cache.get_or_init(|| contract)
313        };
314        let mut contract = template.clone();
315        contract.operation.policy = policy;
316        Ok(contract)
317    }
318
319    // The Contract Of A Built-In Operation Under The Reserved `harmos.` Namespace,
320    // Which Ordinary Declarations Cannot Claim.
321    pub(crate) fn reserved<O: OperationDeclaration>(policy: super::OperationPolicy) -> Result<Self, ContractError> {
322        policy.validate().map_err(crate::Error::from)?;
323        let operation = OperationDefinition {
324            id: crate::Identifier::new(O::ID).map_err(crate::Error::from)?,
325            description: O::DESCRIPTION.to_owned(),
326            policy,
327            visibility: O::VISIBILITY,
328            replay: O::REPLAY,
329            fingerprint: (Entry::of::<O>().fingerprint)()?,
330        };
331        let schema = Schema::from_declarations(
332            &[("request", <O::Request as Described>::DEFINITION), ("response", <O::Response as Described>::DEFINITION)],
333            &SchemaSettings::default(),
334        )
335        .map_err(|source| ContractError::Schema {
336            id: operation.id.to_string(),
337            direction: "contract",
338            source: source.into(),
339        })?;
340        Self::canonical(operation.clone(), operation.fingerprint, Codec::Postcard, schema, &SchemaSettings::default())
341    }
342
343    pub fn operation(&self) -> &OperationDefinition {
344        &self.operation
345    }
346
347    pub fn fingerprint(&self) -> crate::Fingerprint {
348        self.fingerprint
349    }
350
351    pub fn codec(&self) -> Codec {
352        self.codec
353    }
354
355    pub fn schema(&self) -> &Schema {
356        &self.schema
357    }
358
359    pub fn request(&self) -> View<'_> {
360        self.direction("request").expect("validated request root")
361    }
362
363    pub fn response(&self) -> View<'_> {
364        self.direction("response").expect("validated response root")
365    }
366
367    pub fn input(&self) -> Option<View<'_>> {
368        self.direction("input")
369    }
370
371    pub fn output(&self) -> Option<View<'_>> {
372        self.direction("output")
373    }
374
375    pub fn compatible(&self, requirement: &Self) -> bool {
376        self.compatible_with_settings(requirement, &SchemaSettings::default())
377    }
378
379    pub fn compatible_with_settings(&self, requirement: &Self, settings: &SchemaSettings) -> bool {
380        self.operation.id == requirement.operation.id
381            && self.fingerprint == requirement.fingerprint
382            && self.codec == requirement.codec
383            && ["request", "response", "input", "output"].into_iter().all(|name| {
384                match (self.direction(name), requirement.direction(name)) {
385                    (None, None) => true,
386                    (Some(left), Some(right)) => {
387                        self.schema.equivalent(left.index(), &requirement.schema, right.index(), settings).unwrap_or(false)
388                    }
389                    _ => false,
390                }
391            })
392    }
393
394    fn computed_fingerprint(&self) -> Result<crate::Fingerprint, ContractError> {
395        let payload = |view| {
396            schema::identity::fingerprint(view, self.codec.id()).map_err(|source| ContractError::Schema {
397                id: self.operation.id.to_string(),
398                direction: "identity",
399                source: source.into(),
400            })
401        };
402        Ok(schema::identity::operation(
403            self.operation.id.as_str(),
404            payload(self.request())?,
405            payload(self.response())?,
406            self.input().map(payload).transpose()?,
407            self.output().map(payload).transpose()?,
408        ))
409    }
410
411    fn direction(&self, name: &str) -> Option<View<'_>> {
412        let field = self.schema.definitions[self.schema.root].iter().find(|field| field.name == name)?;
413        let Kind::Message { definition } = field.kind else {
414            unreachable!("validated contract direction");
415        };
416        Some(self.schema.view(definition).expect("validated direction reference"))
417    }
418}
419
420impl TryFrom<DecodedContract> for Contract {
421    type Error = ContractError;
422
423    fn try_from(decoded: DecodedContract) -> Result<Self, Self::Error> {
424        if !crate::reserved(decoded.operation.id.as_str()) {
425            return Self::new(decoded.operation, decoded.fingerprint, decoded.codec, decoded.schema)?.with_grpc(decoded.grpc);
426        }
427
428        // Read Only Exact SDK Metadata; Ordinary Declaration Admission Still Rejects These Names.
429        use crate::model::operations::lifecycle::{Action, Descriptor};
430        let refused = || crate::Error::ReservedOperation { id: decoded.operation.id.to_string() }.into();
431        let expected = match decoded.operation.id.as_str() {
432            crate::observers::Control::ID => crate::observers::contract()?,
433            "harmos.initialize" => Descriptor::new(Action::Initialize, decoded.operation.policy).map_err(|_| refused())?.contract().clone(),
434            "harmos.shutdown" => Descriptor::new(Action::Shutdown, decoded.operation.policy).map_err(|_| refused())?.contract().clone(),
435            _ => return Err(refused()),
436        };
437        let identity_matches =
438            decoded.operation == expected.operation && decoded.fingerprint == expected.fingerprint && decoded.codec == expected.codec;
439        let schema_matches = decoded.schema == expected.schema && decoded.grpc == expected.grpc;
440        if !identity_matches || !schema_matches {
441            return Err(refused());
442        }
443        Ok(expected)
444    }
445}
446
447impl Visitor for Contracts {
448    type Entry = WireDeclaration;
449}
450
451impl<M: OperationDeclaration> Member<M> for Contracts {
452    const ENTRY: WireDeclaration = WireDeclaration::of::<M>();
453}
454
455impl Contracts {
456    pub fn of<C: Catalog<Self>>() -> Result<Vec<Contract>, ContractError> {
457        Self::from_projection(&C::ENTRIES)
458    }
459
460    pub fn from_entries(entries: &[WireDeclaration]) -> Result<Vec<Contract>, ContractError> {
461        let mut ids = BTreeSet::new();
462        let mut contracts = Vec::with_capacity(entries.len());
463        for declaration in entries {
464            append(declaration, &mut ids, &mut contracts)?;
465        }
466        Ok(contracts)
467    }
468
469    pub fn from_tree(entries: &[crate::catalog::Node<WireDeclaration>]) -> Result<Vec<Contract>, ContractError> {
470        let mut ids = BTreeSet::new();
471        let mut contracts = Vec::new();
472        crate::catalog::try_for_each(entries, |declaration| append(declaration, &mut ids, &mut contracts))?;
473        Ok(contracts)
474    }
475
476    pub fn from_projection<P: Projection<WireDeclaration>>(entries: &P) -> Result<Vec<Contract>, ContractError> {
477        let mut ids = BTreeSet::new();
478        let mut contracts = Vec::with_capacity(P::LEN);
479        entries.try_for_each(|declaration| append(declaration, &mut ids, &mut contracts))?;
480        Ok(contracts)
481    }
482}
483
484/* === Public Functions === */
485
486/* === Private Functions === */
487
488const fn direction<D: super::Direction>() -> Option<SchemaDefinition> {
489    if D::STREAMED { Some(<D::Item as Described>::DEFINITION) } else { None }
490}
491
492fn append(
493    declaration: &WireDeclaration, ids: &mut BTreeSet<crate::Identifier>, contracts: &mut Vec<Contract>,
494) -> Result<(), ContractError> {
495    let contract = declaration.contract()?;
496    if !ids.insert(contract.operation.id.clone()) {
497        return Err(crate::Error::DuplicateOperation { id: contract.operation.id.to_string() }.into());
498    }
499    contracts.push(contract);
500    Ok(())
501}
502
503/* === Tests === */
504
505#[cfg(test)]
506mod tests {
507    use super::*;
508    use crate::model::operations::lifecycle::{self, Action, Descriptor};
509    use crate::operations::{Backoff, Jitter, RetryPolicy};
510    use serde_json::json;
511
512    fn sdk_contracts() -> [Contract; 3] {
513        let policy = super::super::OperationPolicy {
514            timeout: std::time::Duration::from_secs(17),
515            attempt_timeout: std::time::Duration::from_secs(11),
516            ..lifecycle::POLICY
517        };
518        [
519            crate::observers::contract().unwrap(),
520            Descriptor::new(Action::Initialize, policy).unwrap().contract().clone(),
521            Descriptor::new(Action::Shutdown, policy).unwrap().contract().clone(),
522        ]
523    }
524
525    #[test]
526    fn sdk_metadata_roundtrip() {
527        for original in sdk_contracts() {
528            let encoded = serde_json::to_vec(&original).unwrap();
529            let decoded: Contract = serde_json::from_slice(&encoded).unwrap();
530            assert_eq!(decoded, original);
531        }
532    }
533
534    #[test]
535    fn sdk_metadata_forgery() {
536        let delay = std::time::Duration::from_secs(1);
537        let backoff = Backoff::Fixed { delay, jitter: Jitter::None };
538        for original in sdk_contracts() {
539            let encoded = serde_json::to_value(&original).unwrap();
540            for (pointer, replacement) in [
541                ("/operation/id", json!("harmos.forged")),
542                ("/operation/description", json!("forged SDK contract")),
543                ("/operation/visibility", json!("shared")),
544                ("/operation/policy/attempt_timeout/secs", json!(0)),
545                ("/operation/policy/retries", json!(RetryPolicy::OnTransient { max_retries: 1, backoff })),
546                ("/fingerprint", json!("00".repeat(32))),
547                ("/schema", json!({"root": 0, "definitions": [[]]})),
548            ] {
549                let mut forged = encoded.clone();
550                *forged.pointer_mut(pointer).unwrap() = replacement;
551                assert!(serde_json::from_value::<Contract>(forged).is_err(), "{pointer}");
552            }
553            let mut exposed = encoded;
554            exposed["grpc"] = json!({"package": "forged.v1", "service": "Exposed", "method": "Invoke"});
555            assert!(serde_json::from_value::<Contract>(exposed).is_err());
556        }
557    }
558
559    #[test]
560    fn sdk_lifecycle_refuses_invalid() {
561        let delay = std::time::Duration::from_secs(1);
562        let backoff = Backoff::Fixed { delay, jitter: Jitter::None };
563        let descriptors = [
564            Descriptor::new(Action::Initialize, lifecycle::POLICY).unwrap(),
565            Descriptor::new(Action::Shutdown, lifecycle::POLICY).unwrap(),
566        ];
567        for descriptor in &descriptors {
568            assert_eq!(Contract::from(descriptor.clone()), *descriptor.contract());
569            let encoded = serde_json::to_value(descriptor).unwrap();
570            assert_eq!(serde_json::from_value::<Descriptor>(encoded.clone()).unwrap(), *descriptor);
571            for (pointer, replacement) in [
572                ("/fingerprint", json!("00".repeat(32))),
573                ("/policy/attempt_timeout/secs", json!(0)),
574                ("/policy/retries", json!(RetryPolicy::OnTransient { max_retries: 1, backoff })),
575            ] {
576                let mut forged = encoded.clone();
577                *forged.pointer_mut(pointer).unwrap() = replacement;
578                assert!(serde_json::from_value::<Descriptor>(forged).is_err(), "{pointer}");
579            }
580        }
581        let mut manifest = crate::inventory::contracts::Manifest {
582            facility: "guest".into(),
583            owner: "fixture".into(),
584            provides: Vec::new(),
585            requires: Vec::new(),
586            lifecycle: descriptors,
587        };
588        crate::inventory::contracts::validate(&manifest).unwrap();
589        manifest.lifecycle.swap(0, 1);
590        assert!(crate::inventory::contracts::validate(&manifest).is_err());
591    }
592
593    #[test]
594    fn reserved_operations_refused() {
595        use crate::inventory::contracts::{Manifest, validate};
596        for contract in sdk_contracts() {
597            assert!(contract.operation().validate().is_err());
598            let definition = serde_json::to_value(contract.operation()).unwrap();
599            assert!(serde_json::from_value::<OperationDefinition>(definition).is_err());
600
601            let mut manifest = Manifest {
602                facility: "sidecar".into(),
603                owner: "fixture".into(),
604                provides: Vec::new(),
605                requires: Vec::new(),
606                lifecycle: [
607                    Descriptor::new(Action::Initialize, lifecycle::POLICY).unwrap(),
608                    Descriptor::new(Action::Shutdown, lifecycle::POLICY).unwrap(),
609                ],
610            };
611            validate(&manifest).unwrap();
612            manifest.provides.push(contract);
613            assert!(validate(&manifest).is_err());
614            let encoded = serde_json::to_value(&manifest).unwrap();
615            assert!(serde_json::from_value::<Manifest>(encoded).is_err());
616        }
617    }
618}
harmos_domain/operations/contract.rs

harmos_domain/operations/
contract.rs
