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

harmos_domain/operations/
contract.rs
