lenso_service/production_delivery/
resilience.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use schemars::JsonSchema;
4use serde::{Deserialize, Serialize};
5use utoipa::ToSchema;
6
7use crate::extraction_input_digest;
8
9use super::{
10    ConfigRevision, ContractRetirementEvidence, DeliveryDecision, DeliveryEffects, DeliveryIssue,
11    DeliveryIssueCode, DeploymentObservation, DeploymentPlan, DeploymentReceipt,
12    OperatorObservationAttestation, OperatorObservationAuthorityProvider, PromotionPlan,
13    deployment_observation_integrity_is_valid, deployment_plan_integrity_is_valid, issue,
14    operator_observation_attestation_is_valid, operator_observation_matches_deployment,
15};
16
17pub const COORDINATION_OUTAGE_OBSERVATION_PROTOCOL: &str =
18    "lenso.coordination-outage-observation.v1";
19pub const COORDINATION_OUTAGE_PROTOCOL: &str = "lenso.coordination-outage-proof.v1";
20pub const COORDINATION_RESUME_APPROVAL_PROTOCOL: &str = "lenso.coordination-resume-approval.v1";
21pub const COORDINATION_RESUME_PROTOCOL: &str = "lenso.coordination-resume-receipt.v1";
22
23#[derive(
24    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
25)]
26#[serde(rename_all = "snake_case")]
27pub enum DataPlaneOperation {
28    DirectRequest,
29    Event,
30    DurableWorkflow,
31    Inbox,
32    Outbox,
33    Timer,
34    Retry,
35    Compensation,
36    RuntimeStory,
37}
38
39#[derive(
40    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
41)]
42#[serde(rename_all = "snake_case")]
43pub enum ProtectedCoordinationOperation {
44    Promotion,
45    ConfigurationActivation,
46    ContractRetirement,
47    DeploymentMutation,
48}
49
50/// The exact typed mutation payload protected by a coordination resume approval.
51///
52/// Callers cannot supply a detached digest: both approval and apply derive the
53/// canonical subject digest from this value.
54#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema, ToSchema)]
55#[serde(tag = "operation", content = "subject", rename_all = "snake_case")]
56pub enum CoordinationOperationSubject {
57    Promotion(PromotionPlan),
58    ConfigurationActivation(ConfigRevision),
59    ContractRetirement(ContractRetirementEvidence),
60    DeploymentMutation(DeploymentPlan),
61}
62
63impl CoordinationOperationSubject {
64    #[must_use]
65    pub const fn operation(&self) -> ProtectedCoordinationOperation {
66        match self {
67            Self::Promotion(_) => ProtectedCoordinationOperation::Promotion,
68            Self::ConfigurationActivation(_) => {
69                ProtectedCoordinationOperation::ConfigurationActivation
70            }
71            Self::ContractRetirement(_) => ProtectedCoordinationOperation::ContractRetirement,
72            Self::DeploymentMutation(_) => ProtectedCoordinationOperation::DeploymentMutation,
73        }
74    }
75}
76
77#[must_use]
78pub fn coordination_operation_subject_digest(subject: &CoordinationOperationSubject) -> String {
79    digest_json(&("lenso.coordination-operation-subject.v1", subject))
80}
81
82#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
83#[serde(rename_all = "camelCase")]
84pub struct SecurityContinuity {
85    pub workload_identity_enforced: bool,
86    pub tenant_context_enforced: bool,
87    pub call_policy_enforced: bool,
88    pub service_authorization_enforced: bool,
89}
90
91pub trait CoordinationAuthorityProvider: std::fmt::Debug + Send + Sync {
92    fn sign(&self, authority_id: &str, subject_digest: &str) -> Option<String>;
93
94    fn verify(&self, authority_id: &str, subject_digest: &str, authority_proof: &str) -> bool;
95}
96
97#[derive(Debug, Clone, Default)]
98pub struct DeterministicCoordinationAuthorityProvider {
99    authority_keys: BTreeMap<String, String>,
100}
101
102impl DeterministicCoordinationAuthorityProvider {
103    #[must_use]
104    pub fn new<I, K, V>(authority_keys: I) -> Self
105    where
106        I: IntoIterator<Item = (K, V)>,
107        K: Into<String>,
108        V: Into<String>,
109    {
110        Self {
111            authority_keys: authority_keys
112                .into_iter()
113                .map(|(authority, key)| (authority.into(), key.into()))
114                .collect(),
115        }
116    }
117
118    fn expected_proof(&self, authority_id: &str, subject_digest: &str) -> Option<String> {
119        let key = self.authority_keys.get(authority_id)?;
120        Some(digest_json(&(
121            "lenso.coordination-authority-proof.v1",
122            authority_id,
123            subject_digest,
124            key.as_str(),
125        )))
126    }
127}
128
129impl CoordinationAuthorityProvider for DeterministicCoordinationAuthorityProvider {
130    fn sign(&self, authority_id: &str, subject_digest: &str) -> Option<String> {
131        self.expected_proof(authority_id, subject_digest)
132    }
133
134    fn verify(&self, authority_id: &str, subject_digest: &str, authority_proof: &str) -> bool {
135        self.expected_proof(authority_id, subject_digest).as_deref() == Some(authority_proof)
136    }
137}
138
139#[allow(clippy::struct_excessive_bools)]
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
141#[serde(rename_all = "camelCase")]
142pub struct CoordinationOutageClaims {
143    pub protocol: String,
144    pub deployment_plan_id: String,
145    pub deployment_plan_digest: String,
146    pub deployment_receipt_id: String,
147    pub deployment_observation_id: String,
148    pub operator_observation_id: String,
149    pub operator_observation_digest: String,
150    pub environment_revision_after: u64,
151    pub release_id: String,
152    pub release_digest: String,
153    pub config_revision_id: String,
154    pub system_plane_available: bool,
155    pub runtime_console_available: bool,
156    pub autonomous_service_running: bool,
157    pub selected_gateway_running: bool,
158    pub selected_transport_running: bool,
159    pub gateway_is_data_plane: bool,
160    pub gateway_requires_live_policy: bool,
161    pub gateway_requires_live_release_metadata: bool,
162    pub last_valid_config_revision_available: bool,
163    pub secret_provider_lease_valid: bool,
164    pub secret_rotation_policy_preserved: bool,
165    pub operation_results: BTreeMap<DataPlaneOperation, bool>,
166    pub security: SecurityContinuity,
167    pub durable_checkpoint_id: String,
168    pub evidence_references: Vec<String>,
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
172#[serde(rename_all = "camelCase")]
173pub struct CoordinationOutageObservation {
174    pub protocol: String,
175    pub observation_id: String,
176    pub observation_digest: String,
177    pub authority_id: String,
178    pub authority_proof: String,
179    pub claims: CoordinationOutageClaims,
180}
181
182pub fn attest_coordination_outage(
183    mut claims: CoordinationOutageClaims,
184    authority_id: impl Into<String>,
185    provider: &dyn CoordinationAuthorityProvider,
186) -> Result<CoordinationOutageObservation, DeliveryIssue> {
187    claims.evidence_references.sort();
188    let observation_digest = digest_json(&claims);
189    let authority_id = authority_id.into();
190    let authority_proof = provider
191        .sign(&authority_id, &observation_digest)
192        .ok_or_else(|| {
193            issue(
194                DeliveryIssueCode::CoordinationUnavailable,
195                "The Data Plane observation authority refused to attest the outage window.",
196                "Use the configured outage observation authority at the probe boundary.",
197                "Collect new signed continuity evidence before claiming resilience.",
198            )
199        })?;
200    Ok(CoordinationOutageObservation {
201        protocol: COORDINATION_OUTAGE_OBSERVATION_PROTOCOL.to_owned(),
202        observation_id: format!("coordination-outage-observation:{observation_digest}"),
203        observation_digest,
204        authority_id,
205        authority_proof,
206        claims,
207    })
208}
209
210#[must_use]
211pub fn coordination_outage_observation_integrity_is_valid(
212    observation: &CoordinationOutageObservation,
213    provider: &dyn CoordinationAuthorityProvider,
214) -> bool {
215    observation.protocol == COORDINATION_OUTAGE_OBSERVATION_PROTOCOL
216        && observation.claims.protocol == COORDINATION_OUTAGE_OBSERVATION_PROTOCOL
217        && observation.observation_digest == digest_json(&observation.claims)
218        && observation.observation_id
219            == format!(
220                "coordination-outage-observation:{}",
221                observation.observation_digest
222            )
223        && provider.verify(
224            &observation.authority_id,
225            &observation.observation_digest,
226            &observation.authority_proof,
227        )
228}
229
230#[derive(Debug, Clone)]
231pub struct CoordinationOutageInput {
232    pub deployment_plan: DeploymentPlan,
233    pub deployment: DeploymentReceipt,
234    pub deployment_observation: DeploymentObservation,
235    pub operator_observation: OperatorObservationAttestation,
236    pub outage_observation: CoordinationOutageObservation,
237}
238
239#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
240#[serde(rename_all = "camelCase")]
241pub struct BlockedCoordinationOperation {
242    pub operation: ProtectedCoordinationOperation,
243    pub issue_code: DeliveryIssueCode,
244    pub next_actions: Vec<String>,
245}
246
247#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
248#[serde(rename_all = "camelCase")]
249pub struct CoordinationOutageEvidence {
250    pub protocol: String,
251    pub proof_id: String,
252    pub proof_digest: String,
253    pub decision: DeliveryDecision,
254    pub deployment_plan_id: String,
255    pub deployment_plan_digest: String,
256    pub deployment_receipt_id: String,
257    pub deployment_observation_id: String,
258    pub environment_revision_after: u64,
259    pub release_id: String,
260    pub release_digest: String,
261    pub config_revision_id: String,
262    pub durable_checkpoint_id: String,
263    pub deployment_plan: DeploymentPlan,
264    pub deployment: DeploymentReceipt,
265    pub deployment_observation: DeploymentObservation,
266    pub outage_observation: CoordinationOutageObservation,
267    pub operator_observation: OperatorObservationAttestation,
268    pub continued_operations: Vec<DataPlaneOperation>,
269    pub blocked_operations: Vec<BlockedCoordinationOperation>,
270    pub security: SecurityContinuity,
271    pub issues: Vec<DeliveryIssue>,
272    pub evidence_references: Vec<String>,
273    pub effects: DeliveryEffects,
274}
275
276#[derive(Serialize)]
277#[serde(rename_all = "camelCase")]
278struct CoordinationOutageProofDigestInput<'a> {
279    protocol: &'a str,
280    deployment_plan_id: &'a str,
281    deployment_plan_digest: &'a str,
282    deployment_receipt_id: &'a str,
283    deployment_observation_id: &'a str,
284    environment_revision_after: u64,
285    release_id: &'a str,
286    release_digest: &'a str,
287    config_revision_id: &'a str,
288    durable_checkpoint_id: &'a str,
289    deployment_plan: &'a DeploymentPlan,
290    deployment: &'a DeploymentReceipt,
291    deployment_observation: &'a DeploymentObservation,
292    outage_observation: &'a CoordinationOutageObservation,
293    operator_observation: &'a OperatorObservationAttestation,
294    continued_operations: &'a [DataPlaneOperation],
295    blocked_operations: &'a [BlockedCoordinationOperation],
296    security: &'a SecurityContinuity,
297    issues: &'a [DeliveryIssue],
298    evidence_references: &'a [String],
299    effects: &'a DeliveryEffects,
300}
301
302pub fn prove_system_plane_outage(
303    input: CoordinationOutageInput,
304    outage_provider: &dyn CoordinationAuthorityProvider,
305    operator_provider: &dyn OperatorObservationAuthorityProvider,
306) -> CoordinationOutageEvidence {
307    let claims = &input.outage_observation.claims;
308    let mut issues = Vec::new();
309    if !deployment_plan_integrity_is_valid(&input.deployment_plan)
310        || !deployment_observation_integrity_is_valid(
311            &input.deployment_observation,
312            &input.deployment_plan,
313            &input.deployment,
314        )
315        || !operator_observation_attestation_is_valid(
316            &input.operator_observation,
317            operator_provider,
318        )
319        || !operator_observation_matches_deployment(
320            &input.operator_observation,
321            &input.deployment_plan,
322            &input.deployment,
323            &input.deployment_observation,
324            &input.operator_observation.claims.workload_health,
325        )
326        || !coordination_outage_observation_integrity_is_valid(
327            &input.outage_observation,
328            outage_provider,
329        )
330        || claims.deployment_plan_id != input.deployment_plan.plan_id
331        || claims.deployment_plan_digest != input.deployment_plan.plan_digest
332        || claims.deployment_receipt_id != input.deployment.receipt_id
333        || claims.deployment_observation_id != input.deployment_observation.observation_id
334        || claims.operator_observation_id != input.operator_observation.observation_id
335        || claims.operator_observation_digest != input.operator_observation.observation_digest
336        || claims.environment_revision_after != input.deployment.environment_revision_after
337        || claims.release_id != input.deployment.release_id
338        || claims.release_digest != input.deployment.release_digest
339        || claims.config_revision_id != input.deployment.config_revision_id
340    {
341        issues.push(issue(
342            DeliveryIssueCode::ObservationStale,
343            "Outage evidence is not signed for the exact Deployment plan, receipt, checkpoint, and Operator observation.",
344            "Collect canonical evidence from the trusted Operator and Data Plane probe boundaries.",
345            "Refresh the exact signed observations before proving an outage.",
346        ));
347    }
348    if claims.system_plane_available || claims.runtime_console_available {
349        issues.push(issue(
350            DeliveryIssueCode::CoordinationUnavailable,
351            "The outage proof must actually withhold the System Plane and Console.",
352            "Run the proof with both coordination surfaces unavailable.",
353            "Repeat the outage window without stopping the Autonomous Service.",
354        ));
355    }
356    if !claims.autonomous_service_running
357        || !claims.selected_gateway_running
358        || !claims.selected_transport_running
359        || !claims.gateway_is_data_plane
360        || claims.gateway_requires_live_policy
361        || claims.gateway_requires_live_release_metadata
362    {
363        issues.push(issue(
364            DeliveryIssueCode::CoordinationUnavailable,
365            "Autonomous Service, gateway, or transport continuity still depends on a live coordination surface.",
366            "Keep established traffic and transports entirely inside the Data Plane.",
367            "Correct the topology and repeat the outage proof.",
368        ));
369    }
370    if input.deployment_observation.desired_release_id != input.deployment.release_id
371        || input.deployment_observation.observed_release_id != input.deployment.release_id
372        || input.deployment_observation.drifted
373        || !input.deployment_observation.fresh
374        || input.deployment_observation.config_revision_id != input.deployment.config_revision_id
375    {
376        issues.push(issue(
377            DeliveryIssueCode::ObservationStale,
378            "Outage proof requires a freshly converged production Deployment.",
379            "Converge the exact release and Config Revision before withholding coordination.",
380            "Refresh the Deployment observation and repeat the proof.",
381        ));
382    }
383    if !claims.last_valid_config_revision_available
384        || !claims.secret_provider_lease_valid
385        || !claims.secret_rotation_policy_preserved
386    {
387        issues.push(issue(
388            DeliveryIssueCode::SecretReferenceUnresolved,
389            "Last-valid configuration or declared Secret Provider lease and rotation behavior did not survive the outage.",
390            "Retain only opaque references and honor provider lease and rotation boundaries locally.",
391            "Restore safe secret continuity before claiming Data Plane resilience.",
392        ));
393    }
394    let security_preserved = claims.security.workload_identity_enforced
395        && claims.security.tenant_context_enforced
396        && claims.security.call_policy_enforced
397        && claims.security.service_authorization_enforced;
398    if !security_preserved {
399        issues.push(issue(
400            DeliveryIssueCode::PolicyRuleBlocked,
401            "Coordination loss weakened Workload Identity, Tenant Context, Call Policy, or Service authorization.",
402            "Fail closed rather than bypassing Service-owned security controls.",
403            "Restore the security guarantee and repeat the outage proof.",
404        ));
405    }
406    let expected_operations = expected_data_plane_operations();
407    let observed_operations = claims
408        .operation_results
409        .keys()
410        .copied()
411        .collect::<BTreeSet<_>>();
412    let continued_operations = claims
413        .operation_results
414        .iter()
415        .filter_map(|(operation, continued)| continued.then_some(*operation))
416        .collect::<Vec<_>>();
417    if observed_operations != expected_operations
418        || continued_operations.len() != expected_operations.len()
419    {
420        issues.push(issue(
421            DeliveryIssueCode::ReliabilityEvidenceMissing,
422            "Established requests, events, workflows, inbox, outbox, timers, retries, compensation, or Runtime Story did not all continue.",
423            "Capture every declared Data Plane execution path from the last durable state.",
424            "Repair continuity and repeat the outage proof.",
425        ));
426    }
427    if claims.durable_checkpoint_id.trim().is_empty() {
428        issues.push(issue(
429            DeliveryIssueCode::ReliabilityEvidenceMissing,
430            "Outage evidence lacks a durable resume checkpoint.",
431            "Persist the last valid coordination and Data Plane checkpoint locally.",
432            "Capture durable evidence before restoring coordination.",
433        ));
434    }
435    let blocked_operations = protected_operations();
436    let decision = if issues.is_empty() {
437        DeliveryDecision::Passed
438    } else {
439        DeliveryDecision::Blocked
440    };
441    let effects = DeliveryEffects::default();
442    let proof_digest = digest_json(&CoordinationOutageProofDigestInput {
443        protocol: COORDINATION_OUTAGE_PROTOCOL,
444        deployment_plan_id: input.deployment_plan.plan_id.as_str(),
445        deployment_plan_digest: input.deployment_plan.plan_digest.as_str(),
446        deployment_receipt_id: input.deployment.receipt_id.as_str(),
447        deployment_observation_id: input.deployment_observation.observation_id.as_str(),
448        environment_revision_after: input.deployment.environment_revision_after,
449        release_id: input.deployment.release_id.as_str(),
450        release_digest: input.deployment.release_digest.as_str(),
451        config_revision_id: input.deployment.config_revision_id.as_str(),
452        durable_checkpoint_id: claims.durable_checkpoint_id.as_str(),
453        deployment_plan: &input.deployment_plan,
454        deployment: &input.deployment,
455        deployment_observation: &input.deployment_observation,
456        outage_observation: &input.outage_observation,
457        operator_observation: &input.operator_observation,
458        continued_operations: continued_operations.as_slice(),
459        blocked_operations: blocked_operations.as_slice(),
460        security: &claims.security,
461        issues: issues.as_slice(),
462        evidence_references: claims.evidence_references.as_slice(),
463        effects: &effects,
464    });
465    CoordinationOutageEvidence {
466        protocol: COORDINATION_OUTAGE_PROTOCOL.to_owned(),
467        proof_id: format!("coordination-outage-proof:{proof_digest}"),
468        proof_digest,
469        decision,
470        deployment_plan_id: input.deployment_plan.plan_id.clone(),
471        deployment_plan_digest: input.deployment_plan.plan_digest.clone(),
472        deployment_receipt_id: input.deployment.receipt_id.clone(),
473        deployment_observation_id: input.deployment_observation.observation_id.clone(),
474        environment_revision_after: input.deployment.environment_revision_after,
475        release_id: input.deployment.release_id.clone(),
476        release_digest: input.deployment.release_digest.clone(),
477        config_revision_id: input.deployment.config_revision_id.clone(),
478        durable_checkpoint_id: claims.durable_checkpoint_id.clone(),
479        deployment_plan: input.deployment_plan.clone(),
480        deployment: input.deployment.clone(),
481        deployment_observation: input.deployment_observation.clone(),
482        outage_observation: input.outage_observation.clone(),
483        operator_observation: input.operator_observation.clone(),
484        continued_operations,
485        blocked_operations,
486        security: claims.security.clone(),
487        issues,
488        evidence_references: claims.evidence_references.clone(),
489        effects,
490    }
491}
492
493#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
494#[serde(rename_all = "camelCase")]
495pub struct CoordinationResumeApproval {
496    pub protocol: String,
497    pub approval_id: String,
498    pub approval_digest: String,
499    pub operation_id: String,
500    pub operation: ProtectedCoordinationOperation,
501    pub operation_subject_digest: String,
502    pub outage_proof_id: String,
503    pub outage_observation_id: String,
504    pub deployment_receipt_id: String,
505    pub durable_checkpoint_id: String,
506    pub environment_revision_after: u64,
507    pub coordination_revision: u64,
508    pub authority_id: String,
509    pub authority_proof: String,
510}
511
512#[allow(clippy::too_many_arguments)]
513pub fn approve_coordination_resume(
514    proof: &CoordinationOutageEvidence,
515    operation_id: impl Into<String>,
516    subject: &CoordinationOperationSubject,
517    coordination_revision: u64,
518    authority_id: impl Into<String>,
519    outage_provider: &dyn CoordinationAuthorityProvider,
520    operator_provider: &dyn OperatorObservationAuthorityProvider,
521    approval_provider: &dyn CoordinationAuthorityProvider,
522) -> Result<CoordinationResumeApproval, Vec<DeliveryIssue>> {
523    let operation_id = operation_id.into();
524    let operation = subject.operation();
525    let operation_subject_digest = coordination_operation_subject_digest(subject);
526    let authority_id = authority_id.into();
527    if !coordination_outage_evidence_integrity_is_valid(proof, outage_provider, operator_provider)
528        || proof.decision != DeliveryDecision::Passed
529        || operation_id.trim().is_empty()
530        || coordination_revision <= proof.environment_revision_after
531        || !proof
532            .blocked_operations
533            .iter()
534            .any(|blocked| blocked.operation == operation)
535    {
536        return Err(vec![coordination_resume_issue()]);
537    }
538    let approval_digest = coordination_resume_approval_digest(
539        &operation_id,
540        operation,
541        &operation_subject_digest,
542        proof,
543        coordination_revision,
544        &authority_id,
545    );
546    let Some(authority_proof) = approval_provider.sign(&authority_id, &approval_digest) else {
547        return Err(vec![coordination_resume_issue()]);
548    };
549    Ok(CoordinationResumeApproval {
550        protocol: COORDINATION_RESUME_APPROVAL_PROTOCOL.to_owned(),
551        approval_id: format!("coordination-resume-approval:{approval_digest}"),
552        approval_digest,
553        operation_id,
554        operation,
555        operation_subject_digest,
556        outage_proof_id: proof.proof_id.clone(),
557        outage_observation_id: proof.outage_observation.observation_id.clone(),
558        deployment_receipt_id: proof.deployment_receipt_id.clone(),
559        durable_checkpoint_id: proof.durable_checkpoint_id.clone(),
560        environment_revision_after: proof.environment_revision_after,
561        coordination_revision,
562        authority_id,
563        authority_proof,
564    })
565}
566
567#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
568#[serde(rename_all = "camelCase")]
569/// A deterministic authorization receipt for retrying the protected operation.
570///
571/// This receipt deliberately records no environment, deployment, configuration,
572/// or ledger effect. The operation named by `operation_id` must execute through
573/// its own durable idempotency boundary after this authorization is obtained.
574pub struct CoordinationResumeReceipt {
575    pub protocol: String,
576    pub receipt_id: String,
577    pub operation_id: String,
578    pub operation: ProtectedCoordinationOperation,
579    pub operation_subject_digest: String,
580    pub outage_proof_id: String,
581    pub durable_checkpoint_id: String,
582    pub approval_id: String,
583    pub approval_digest: String,
584    pub coordination_revision: u64,
585    pub effects: DeliveryEffects,
586}
587
588#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
589#[serde(rename_all = "camelCase")]
590pub struct CoordinationResumeState {
591    #[serde(default)]
592    pub receipts: Vec<CoordinationResumeReceipt>,
593}
594
595#[allow(clippy::too_many_arguments)]
596pub fn resume_protected_operation(
597    state: &mut CoordinationResumeState,
598    proof: &CoordinationOutageEvidence,
599    approval: &CoordinationResumeApproval,
600    subject: &CoordinationOperationSubject,
601    current_coordination_revision: u64,
602    outage_provider: &dyn CoordinationAuthorityProvider,
603    operator_provider: &dyn OperatorObservationAuthorityProvider,
604    approval_provider: &dyn CoordinationAuthorityProvider,
605) -> Result<CoordinationResumeReceipt, Vec<DeliveryIssue>> {
606    if !coordination_outage_evidence_integrity_is_valid(proof, outage_provider, operator_provider)
607        || proof.decision != DeliveryDecision::Passed
608        || !coordination_resume_approval_integrity_is_valid(approval, proof, approval_provider)
609        || subject.operation() != approval.operation
610        || coordination_operation_subject_digest(subject) != approval.operation_subject_digest
611        || current_coordination_revision != approval.coordination_revision
612    {
613        return Err(vec![coordination_resume_issue()]);
614    }
615    if let Some(receipt) = state
616        .receipts
617        .iter()
618        .find(|receipt| receipt.operation_id == approval.operation_id)
619    {
620        return coordination_resume_receipt_integrity_is_valid(receipt, proof, approval)
621            .then(|| receipt.clone())
622            .ok_or_else(|| {
623                vec![issue(
624                    DeliveryIssueCode::StaleInput,
625                    "The resume operation identity was reused with different protected evidence.",
626                    "Preserve the exact operation, outage proof, durable checkpoint, and approval for idempotent replay.",
627                    "Use the original request or allocate a new stable operation identifier.",
628                )]
629            });
630    }
631    let effects = coordination_resume_authorization_effects();
632    let receipt_id = format!(
633        "coordination-resume-receipt:{}",
634        digest_json(&(
635            COORDINATION_RESUME_PROTOCOL,
636            approval.operation_id.as_str(),
637            approval.operation,
638            approval.operation_subject_digest.as_str(),
639            proof.proof_id.as_str(),
640            proof.durable_checkpoint_id.as_str(),
641            approval.approval_id.as_str(),
642            approval.approval_digest.as_str(),
643            approval.coordination_revision,
644            &effects,
645        ))
646    );
647    let receipt = CoordinationResumeReceipt {
648        protocol: COORDINATION_RESUME_PROTOCOL.to_owned(),
649        receipt_id,
650        operation_id: approval.operation_id.clone(),
651        operation: approval.operation,
652        operation_subject_digest: approval.operation_subject_digest.clone(),
653        outage_proof_id: proof.proof_id.clone(),
654        durable_checkpoint_id: proof.durable_checkpoint_id.clone(),
655        approval_id: approval.approval_id.clone(),
656        approval_digest: approval.approval_digest.clone(),
657        coordination_revision: approval.coordination_revision,
658        effects,
659    };
660    state.receipts.push(receipt.clone());
661    Ok(receipt)
662}
663
664#[must_use]
665pub fn coordination_outage_evidence_integrity_is_valid(
666    proof: &CoordinationOutageEvidence,
667    outage_provider: &dyn CoordinationAuthorityProvider,
668    operator_provider: &dyn OperatorObservationAuthorityProvider,
669) -> bool {
670    let expected = prove_system_plane_outage(
671        CoordinationOutageInput {
672            deployment_plan: proof.deployment_plan.clone(),
673            deployment: proof.deployment.clone(),
674            deployment_observation: proof.deployment_observation.clone(),
675            operator_observation: proof.operator_observation.clone(),
676            outage_observation: proof.outage_observation.clone(),
677        },
678        outage_provider,
679        operator_provider,
680    );
681    *proof == expected
682}
683
684/// Recomputes the content digest of an outage proof without conferring authority.
685/// Consumers must still call [`coordination_outage_evidence_integrity_is_valid`],
686/// which deterministically replays the signed evidence semantics.
687#[must_use]
688pub fn coordination_outage_evidence_digest(proof: &CoordinationOutageEvidence) -> String {
689    digest_json(&CoordinationOutageProofDigestInput {
690        protocol: proof.protocol.as_str(),
691        deployment_plan_id: proof.deployment_plan_id.as_str(),
692        deployment_plan_digest: proof.deployment_plan_digest.as_str(),
693        deployment_receipt_id: proof.deployment_receipt_id.as_str(),
694        deployment_observation_id: proof.deployment_observation_id.as_str(),
695        environment_revision_after: proof.environment_revision_after,
696        release_id: proof.release_id.as_str(),
697        release_digest: proof.release_digest.as_str(),
698        config_revision_id: proof.config_revision_id.as_str(),
699        durable_checkpoint_id: proof.durable_checkpoint_id.as_str(),
700        deployment_plan: &proof.deployment_plan,
701        deployment: &proof.deployment,
702        deployment_observation: &proof.deployment_observation,
703        outage_observation: &proof.outage_observation,
704        operator_observation: &proof.operator_observation,
705        continued_operations: proof.continued_operations.as_slice(),
706        blocked_operations: proof.blocked_operations.as_slice(),
707        security: &proof.security,
708        issues: proof.issues.as_slice(),
709        evidence_references: proof.evidence_references.as_slice(),
710        effects: &proof.effects,
711    })
712}
713
714#[must_use]
715pub fn coordination_resume_approval_integrity_is_valid(
716    approval: &CoordinationResumeApproval,
717    proof: &CoordinationOutageEvidence,
718    provider: &dyn CoordinationAuthorityProvider,
719) -> bool {
720    approval.protocol == COORDINATION_RESUME_APPROVAL_PROTOCOL
721        && approval.approval_digest
722            == coordination_resume_approval_digest(
723                &approval.operation_id,
724                approval.operation,
725                &approval.operation_subject_digest,
726                proof,
727                approval.coordination_revision,
728                &approval.authority_id,
729            )
730        && approval.approval_id
731            == format!("coordination-resume-approval:{}", approval.approval_digest)
732        && approval.outage_proof_id == proof.proof_id
733        && approval.outage_observation_id == proof.outage_observation.observation_id
734        && approval.deployment_receipt_id == proof.deployment_receipt_id
735        && approval.durable_checkpoint_id == proof.durable_checkpoint_id
736        && approval.environment_revision_after == proof.environment_revision_after
737        && approval.coordination_revision > proof.environment_revision_after
738        && !approval.operation_id.trim().is_empty()
739        && valid_operation_subject_digest(&approval.operation_subject_digest)
740        && proof
741            .blocked_operations
742            .iter()
743            .any(|blocked| blocked.operation == approval.operation)
744        && provider.verify(
745            &approval.authority_id,
746            &approval.approval_digest,
747            &approval.authority_proof,
748        )
749}
750
751#[must_use]
752pub fn coordination_resume_receipt_integrity_is_valid(
753    receipt: &CoordinationResumeReceipt,
754    proof: &CoordinationOutageEvidence,
755    approval: &CoordinationResumeApproval,
756) -> bool {
757    let effects = coordination_resume_authorization_effects();
758    receipt.protocol == COORDINATION_RESUME_PROTOCOL
759        && receipt.operation_id == approval.operation_id
760        && receipt.operation == approval.operation
761        && receipt.operation_subject_digest == approval.operation_subject_digest
762        && receipt.outage_proof_id == proof.proof_id
763        && receipt.durable_checkpoint_id == proof.durable_checkpoint_id
764        && receipt.approval_id == approval.approval_id
765        && receipt.approval_digest == approval.approval_digest
766        && receipt.coordination_revision == approval.coordination_revision
767        && receipt.effects == effects
768        && receipt.receipt_id
769            == format!(
770                "coordination-resume-receipt:{}",
771                digest_json(&(
772                    receipt.protocol.as_str(),
773                    receipt.operation_id.as_str(),
774                    receipt.operation,
775                    receipt.operation_subject_digest.as_str(),
776                    receipt.outage_proof_id.as_str(),
777                    receipt.durable_checkpoint_id.as_str(),
778                    receipt.approval_id.as_str(),
779                    receipt.approval_digest.as_str(),
780                    receipt.coordination_revision,
781                    &receipt.effects,
782                ))
783            )
784}
785
786fn coordination_resume_approval_digest(
787    operation_id: &str,
788    operation: ProtectedCoordinationOperation,
789    operation_subject_digest: &str,
790    proof: &CoordinationOutageEvidence,
791    coordination_revision: u64,
792    authority_id: &str,
793) -> String {
794    digest_json(&(
795        COORDINATION_RESUME_APPROVAL_PROTOCOL,
796        operation_id,
797        operation,
798        operation_subject_digest,
799        proof.proof_id.as_str(),
800        proof.outage_observation.observation_id.as_str(),
801        proof.deployment_receipt_id.as_str(),
802        proof.durable_checkpoint_id.as_str(),
803        proof.environment_revision_after,
804        coordination_revision,
805        authority_id,
806    ))
807}
808
809fn valid_operation_subject_digest(digest: &str) -> bool {
810    digest.strip_prefix("sha256:").is_some_and(|value| {
811        value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit())
812    })
813}
814
815fn expected_data_plane_operations() -> BTreeSet<DataPlaneOperation> {
816    [
817        DataPlaneOperation::DirectRequest,
818        DataPlaneOperation::Event,
819        DataPlaneOperation::DurableWorkflow,
820        DataPlaneOperation::Inbox,
821        DataPlaneOperation::Outbox,
822        DataPlaneOperation::Timer,
823        DataPlaneOperation::Retry,
824        DataPlaneOperation::Compensation,
825        DataPlaneOperation::RuntimeStory,
826    ]
827    .into_iter()
828    .collect()
829}
830
831fn protected_operations() -> Vec<BlockedCoordinationOperation> {
832    [
833        ProtectedCoordinationOperation::Promotion,
834        ProtectedCoordinationOperation::ConfigurationActivation,
835        ProtectedCoordinationOperation::ContractRetirement,
836        ProtectedCoordinationOperation::DeploymentMutation,
837    ]
838    .into_iter()
839    .map(|operation| BlockedCoordinationOperation {
840        operation,
841        issue_code: DeliveryIssueCode::CoordinationUnavailable,
842        next_actions: vec![
843            "Restore coordination and refresh policy, approval, and environment evidence."
844                .to_owned(),
845            "Resume from the durable checkpoint without replaying completed effects.".to_owned(),
846        ],
847    })
848    .collect()
849}
850
851fn coordination_resume_issue() -> DeliveryIssue {
852    issue(
853        DeliveryIssueCode::CoordinationUnavailable,
854        "Protected coordination cannot resume without signed outage evidence, a newer coordination revision, and typed approval for the exact operation.",
855        "Restore coordination and refresh the exact signed operation approval boundary.",
856        "Retry the same stable operation identifier with current authority evidence.",
857    )
858}
859
860fn coordination_resume_authorization_effects() -> DeliveryEffects {
861    DeliveryEffects::default()
862}
863
864fn digest_json(value: &impl Serialize) -> String {
865    extraction_input_digest(serde_json::to_vec(value).expect("resilience values must serialize"))
866}
lenso_service/production_delivery/resilience.rs

lenso_service/production_delivery/
resilience.rs
