history.rs (7516B)
1 //! Bounded projections of backend-owned delivery claim receipts. 2 3 use super::{AuthoredDeliveryPlan, DELIVERY_PLAN_ATTEMPTS_MAX, WorkClaim}; 4 use crate::{ 5 Error, 6 authored_atomic::{ 7 AuthoredAtomicCommand, AuthoredAtomicOutcome, AuthoredAtomicReceipt, ClaimAuthoredTarget, 8 ClaimAuthoredWork, 9 }, 10 }; 11 12 /// One issued claim and its original scheduling-attempt boundary. 13 #[derive(Clone, Debug, Eq, PartialEq)] 14 pub struct AuthoredDeliveryClaim { 15 claim: WorkClaim, 16 prior_attempt_count: u32, 17 } 18 19 impl AuthoredDeliveryClaim { 20 pub const fn claim(&self) -> &WorkClaim { 21 &self.claim 22 } 23 pub const fn prior_attempt_count(&self) -> u32 { 24 self.prior_attempt_count 25 } 26 } 27 28 /// A consistent, bounded read of a plan and its immutable issued claims. 29 /// 30 /// Backends must supply their own retained receipts from the same read snapshot. 31 /// A truncated or unproven legacy history never proves absence of an attempt. 32 #[derive(Clone, Debug, Eq, PartialEq)] 33 pub struct AuthoredDeliveryHistory { 34 plan: AuthoredDeliveryPlan, 35 claims: Vec<AuthoredDeliveryClaim>, 36 initial_boundary_proven: bool, 37 truncated: bool, 38 } 39 40 impl AuthoredDeliveryHistory { 41 pub fn new( 42 plan: AuthoredDeliveryPlan, 43 preparation: Option<&AuthoredAtomicReceipt>, 44 ) -> Result<Self, Error> { 45 plan.validate()?; 46 let initial_boundary_proven = match preparation.map(AuthoredAtomicReceipt::outcome) { 47 Some(AuthoredAtomicOutcome::Prepared { delivery_plans, .. }) => { 48 Self::initial_boundary(&plan, delivery_plans)? 49 } 50 Some(AuthoredAtomicOutcome::Submitted(value)) => { 51 value.validate()?; 52 Self::initial_boundary(&plan, value.preparation().delivery_plans())? 53 } 54 Some(_) => return Err(Error::AtomicWorkflowMismatch), 55 None => false, 56 }; 57 Ok(Self { 58 plan, 59 claims: Vec::new(), 60 initial_boundary_proven, 61 truncated: false, 62 }) 63 } 64 65 fn initial_boundary( 66 plan: &AuthoredDeliveryPlan, 67 originals: &[AuthoredDeliveryPlan], 68 ) -> Result<bool, Error> { 69 let original = originals 70 .iter() 71 .find(|original| original.plan_id() == plan.plan_id()) 72 .ok_or(Error::AtomicWorkflowMismatch)?; 73 original.validate()?; 74 if original.artifact_id() != plan.artifact_id() 75 || original.intent() != plan.intent() 76 || original.created_at_unix_ms() != plan.created_at_unix_ms() 77 || original.revision() > plan.revision() 78 || original.updated_at_unix_ms() > plan.updated_at_unix_ms() 79 { 80 return Err(Error::AtomicWorkflowMismatch); 81 } 82 Ok(original.claim_evidence().is_none() && original.attempts().is_empty()) 83 } 84 85 pub fn push_claim(&mut self, receipt: &AuthoredAtomicReceipt) -> Result<(), Error> { 86 if self.claims.len() >= DELIVERY_PLAN_ATTEMPTS_MAX as usize { 87 return Err(Error::DeliveryAttemptOverflow); 88 } 89 let AuthoredAtomicOutcome::DeliveryPlan(original) = receipt.outcome() else { 90 return Err(Error::AtomicWorkflowMismatch); 91 }; 92 original.validate()?; 93 let claim = original 94 .claim_evidence() 95 .ok_or(Error::AtomicWorkflowMismatch)?; 96 let command = AuthoredAtomicCommand::Claim(ClaimAuthoredWork::new( 97 ClaimAuthoredTarget::DeliveryPlan(self.plan.plan_id()), 98 claim.clone(), 99 )); 100 if !receipt.matches_command(&command) 101 || receipt.committed_at_unix_ms() != claim.acquired_at_unix_ms() 102 || original.plan_id() != self.plan.plan_id() 103 || original.artifact_id() != self.plan.artifact_id() 104 || original.request().is_none() 105 || original.request() != self.plan.request() 106 || original.intent() != self.plan.intent() 107 || original.created_at_unix_ms() != self.plan.created_at_unix_ms() 108 || original.revision() > self.plan.revision() 109 || original.updated_at_unix_ms() > self.plan.updated_at_unix_ms() 110 || self.claims.iter().any(|entry| entry.claim == *claim) 111 { 112 return Err(Error::AtomicWorkflowMismatch); 113 } 114 self.claims.push(AuthoredDeliveryClaim { 115 claim: claim.clone(), 116 prior_attempt_count: original.attempt_count(), 117 }); 118 Ok(()) 119 } 120 121 /// Records that additional retained claims exceeded this bounded read. 122 pub fn mark_truncated(&mut self) { 123 self.truncated = true; 124 } 125 pub const fn plan(&self) -> &AuthoredDeliveryPlan { 126 &self.plan 127 } 128 pub fn claims(&self) -> &[AuthoredDeliveryClaim] { 129 &self.claims 130 } 131 pub const fn is_complete(&self) -> bool { 132 self.initial_boundary_proven && !self.truncated 133 } 134 pub const fn is_truncated(&self) -> bool { 135 self.truncated 136 } 137 pub fn validate(&self) -> Result<(), Error> { 138 self.plan.validate()?; 139 if self.is_complete() 140 && (self 141 .plan 142 .claim_evidence() 143 .is_some_and(|claim| !self.claims.iter().any(|entry| entry.claim() == claim)) 144 || self.plan.delivery_facts().iter().any(|fact| { 145 !self 146 .claims 147 .iter() 148 .any(|entry| entry.claim() == fact.claim()) 149 })) 150 { 151 return Err(Error::AtomicWorkflowMismatch); 152 } 153 Ok(()) 154 } 155 /// Require original backend claim provenance before scheduling reconciliation. 156 pub fn require_pending_fact_provenance(&self) -> Result<(), Error> { 157 self.validate()?; 158 if self.is_truncated() { 159 return Err(Error::DeliveryAttemptOverflow); 160 } 161 if self.plan.pending_delivery_facts().any(|fact| { 162 !self 163 .claims 164 .iter() 165 .any(|entry| entry.claim() == fact.claim()) 166 }) { 167 return Err(Error::AtomicWorkflowMismatch); 168 } 169 Ok(()) 170 } 171 pub fn proves_no_issued_attempt(&self) -> bool { 172 self.is_complete() 173 && self.claims.is_empty() 174 && self.plan.attempts().is_empty() 175 && self.plan.delivery_facts().is_empty() 176 && self.plan.claim_evidence().is_none() 177 } 178 179 pub fn has_unresolved_claims(&self) -> bool { 180 !self.is_complete() 181 || self.claims.iter().any(|entry| { 182 !self 183 .plan 184 .delivery_facts() 185 .iter() 186 .any(|fact| fact.claim() == &entry.claim) 187 && !self.plan.attempts().iter().any(|attempt| { 188 // Historical fenced applications have no explicit claim 189 // marker. Their original attempt boundary and valid lease 190 // interval together identify the retained application. 191 attempt.claim_evidence().is_none() 192 && entry.prior_attempt_count.checked_add(1) 193 == Some(attempt.attempt().get()) 194 && attempt.recorded_at_unix_ms() >= entry.claim.acquired_at_unix_ms() 195 && attempt.recorded_at_unix_ms() < entry.claim.expires_at_unix_ms() 196 }) 197 }) 198 } 199 }