ic_backup/model/attempt_journal/
mod.rs

1//! Durable finite local attempt accounting adapted from Canic's pending/receipt contracts.
2
3mod authority;
4mod history;
5pub use authority::{
6    AttemptAuthorityRecord, AttemptBudgetRecord, MAX_OPERATION_ATTEMPTS, OperationBindingRecord,
7    OperationBindingRequest,
8};
9
10use crate::model::artifacts::{ArtifactChecksumRecord, ChecksumError, canonical_hash};
11use history::{AttemptEventRecord, Projection};
12use serde::{Deserialize, Deserializer, Serialize, de};
13use std::fmt;
14use thiserror::Error;
15
16/// Maximum retained reservation/receipt events in one operation journal.
17pub const MAX_ATTEMPT_EVENTS: usize = 2048;
18/// Maximum encoded input and canonical output bytes admitted by persistence ops.
19pub const MAX_ATTEMPT_JOURNAL_BYTES: u64 = 1024 * 1024;
20
21/// Integration-qualified direct mutation outcome; an invocation failure is insufficient.
22#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
23#[serde(rename_all = "snake_case")]
24pub enum MutationOutcomeRecord {
25    /// Exact requested effect is qualified as applied.
26    Applied,
27    /// Exact requested effect is qualified as not applied; consumed allowance remains spent.
28    NotApplied,
29}
30
31/// Integration-qualified observation outcome for its exact reserved mutation.
32#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
33#[serde(rename_all = "snake_case")]
34pub enum ObservationOutcomeRecord {
35    /// Observation qualifies exact requested effect as applied.
36    Applied,
37    /// Observation qualifies exact requested effect as not applied.
38    NotApplied,
39    /// Settled observation cannot resolve the mutation; no allowance is restored.
40    Uncertain,
41}
42
43/// Passive exact direct-reply evidence supplied by its qualified owner.
44#[derive(Clone, Debug)]
45pub struct MutationReceiptRequest {
46    /// Previously reserved mutation attempt number.
47    pub attempt: u32,
48    /// Exact mutating request digest, checked against immutable authority.
49    pub request: String,
50    /// Qualified effect outcome; local process exit alone cannot produce it.
51    pub outcome: MutationOutcomeRecord,
52    /// Canonical digest of retained owner evidence.
53    pub evidence: String,
54}
55
56/// Passive exact reconciliation-reply evidence supplied by its qualified owner.
57///
58/// The owner qualifies observation completion, custody and paid-effect settlement.
59/// A lost observation reply stays pending; it cannot be recorded as `Uncertain`
60/// merely because the local invocation failed or its response was lost.
61#[derive(Clone, Debug)]
62pub struct ObservationReceiptRequest {
63    /// Previously reserved observation attempt number.
64    pub attempt: u32,
65    /// Exact observation request digest, checked against its reservation.
66    pub request: String,
67    /// Qualified outcome, including explicit unresolved evidence.
68    pub outcome: ObservationOutcomeRecord,
69    /// Canonical digest of retained owner evidence.
70    pub evidence: String,
71}
72
73/// Read-only derived local accounting; does not schedule or authorize effects.
74#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
75pub struct AttemptJournalView {
76    /// Exact immutable authority digest including original budgets.
77    pub authority: ArtifactChecksumRecord,
78    /// Mutation attempts already consumed, without refunds.
79    pub mutations_used: u32,
80    /// Reconciliation observations already consumed, without refunds.
81    pub observations_used: u32,
82    /// Original mutation allowance still unconsumed.
83    pub mutations_remaining: u32,
84    /// Original observation allowance still unconsumed.
85    pub observations_remaining: u32,
86    /// Exact unresolved mutation attempt, if any.
87    pub pending_mutation: Option<u32>,
88    /// Exact unresolved observation attempt, if any.
89    pub pending_observation: Option<u32>,
90    /// Retained qualified application of this operation, not full run completion.
91    pub applied: bool,
92    /// Number of retained reservation and receipt events.
93    pub history_len: usize,
94}
95
96/// Maintained v1 append-only local reservations and receipts for one exact operation.
97#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
98#[serde(try_from = "JournalFields")]
99pub struct AttemptJournalRecord {
100    version: u16,
101    authority: AttemptAuthorityRecord,
102    events: Vec<AttemptEventRecord>,
103    #[serde(skip)]
104    projection: Projection,
105}
106
107#[derive(Deserialize)]
108#[serde(deny_unknown_fields)]
109struct JournalFields {
110    version: u16,
111    authority: AttemptAuthorityRecord,
112    #[serde(deserialize_with = "bounded_events")]
113    events: Vec<AttemptEventRecord>,
114}
115
116impl TryFrom<JournalFields> for AttemptJournalRecord {
117    type Error = AttemptJournalRecordError;
118    fn try_from(mut fields: JournalFields) -> Result<Self, Self::Error> {
119        if fields.version != 1 {
120            return Err(AttemptJournalRecordError::UnsupportedVersion(
121                fields.version,
122            ));
123        }
124        let mut projection = Projection::empty();
125        for event in &mut fields.events {
126            event.normalize()?;
127            projection.apply(event, fields.authority.budget())?;
128        }
129        Ok(Self {
130            version: 1,
131            authority: fields.authority,
132            events: fields.events,
133            projection,
134        })
135    }
136}
137
138impl AttemptJournalRecord {
139    /// Retain an exact validated authority with no consumed attempts.
140    #[must_use]
141    pub const fn new(authority: AttemptAuthorityRecord) -> Self {
142        Self {
143            version: 1,
144            authority,
145            events: Vec::new(),
146            projection: Projection::empty(),
147        }
148    }
149    /// Read exact declared identity and immutable ceilings.
150    #[must_use]
151    pub const fn authority(&self) -> &AttemptAuthorityRecord {
152        &self.authority
153    }
154    /// Read the exact canonical request of the unresolved observation, if any.
155    ///
156    /// This projects existing validated history. It creates no reservation or
157    /// repeat-dispatch permission and returns None after a settled observation.
158    #[must_use]
159    pub fn pending_observation_request(&self) -> Option<&str> {
160        self.projection
161            .pending_observation
162            .as_ref()
163            .map(|pending| pending.request.as_str())
164    }
165    /// Project local evidence without IO or replenishing authority.
166    ///
167    /// This excludes individual receipt evidence; use `Self::digest` for exact history identity.
168    #[must_use]
169    pub fn view(&self) -> AttemptJournalView {
170        AttemptJournalView {
171            authority: self.authority.digest(),
172            mutations_used: self.projection.mutations_used,
173            observations_used: self.projection.observations_used,
174            mutations_remaining: self.authority.budget().mutations()
175                - self.projection.mutations_used,
176            observations_remaining: self.authority.budget().observations()
177                - self.projection.observations_used,
178            pending_mutation: self.projection.pending_mutation,
179            pending_observation: self
180                .projection
181                .pending_observation
182                .as_ref()
183                .map(|pending| pending.attempt),
184            applied: self.projection.applied,
185            history_len: self.events.len(),
186        }
187    }
188    /// Hash full original authority and every canonical chronological reservation/receipt.
189    ///
190    /// The NUL-terminated v1 domain precedes 64 ASCII authority hash bytes, a
191    /// big-endian u64 event count and closed tagged event bytes. Derived projections,
192    /// JSON formatting and filesystem paths are excluded. This authenticates no receipt.
193    #[must_use]
194    pub fn digest(&self) -> ArtifactChecksumRecord {
195        let mut bytes = b"ic-backup/attempt-journal-history/v1\0".to_vec();
196        bytes.extend_from_slice(self.authority.digest().hash().as_bytes());
197        bytes.extend_from_slice(&(self.events.len() as u64).to_be_bytes());
198        for event in &self.events {
199            event.append_digest_bytes(&mut bytes);
200        }
201        ArtifactChecksumRecord::from_bytes(&bytes)
202    }
203    pub(crate) fn reserve_mutation(&mut self) -> Result<u32, AttemptJournalRecordError> {
204        let attempt = self.projection.next_attempt;
205        self.append(AttemptEventRecord::MutationReserved { attempt })?;
206        Ok(attempt)
207    }
208    pub(crate) fn reserve_observation(
209        &mut self,
210        mutation: u32,
211        request: &str,
212    ) -> Result<u32, AttemptJournalRecordError> {
213        let attempt = self.projection.next_attempt;
214        self.append(AttemptEventRecord::ObservationReserved {
215            attempt,
216            mutation,
217            request: request.to_owned(),
218        })?;
219        Ok(attempt)
220    }
221    pub(crate) fn record_mutation(
222        &mut self,
223        receipt: MutationReceiptRequest,
224    ) -> Result<(), AttemptJournalRecordError> {
225        if canonical_hash(&receipt.request)? != self.authority.binding().request() {
226            return Err(AttemptJournalRecordError::RequestMismatch);
227        }
228        self.append(AttemptEventRecord::MutationResolved {
229            mutation: receipt.attempt,
230            outcome: receipt.outcome,
231            evidence: receipt.evidence,
232        })
233    }
234    pub(crate) fn record_observation(
235        &mut self,
236        receipt: ObservationReceiptRequest,
237    ) -> Result<(), AttemptJournalRecordError> {
238        self.append(AttemptEventRecord::ObservationRecorded {
239            observation: receipt.attempt,
240            request: receipt.request,
241            outcome: receipt.outcome,
242            evidence: receipt.evidence,
243        })
244    }
245    fn append(&mut self, mut event: AttemptEventRecord) -> Result<(), AttemptJournalRecordError> {
246        if self.events.len() == MAX_ATTEMPT_EVENTS {
247            return Err(AttemptJournalRecordError::HistoryTooLarge);
248        }
249        event.normalize()?;
250        let mut projection = self.projection.clone();
251        projection.apply(&event, self.authority.budget())?;
252        self.events.push(event);
253        self.projection = projection;
254        Ok(())
255    }
256}
257
258fn bounded_events<'de, D: Deserializer<'de>>(
259    deserializer: D,
260) -> Result<Vec<AttemptEventRecord>, D::Error> {
261    struct EventsVisitor;
262    impl<'de> de::Visitor<'de> for EventsVisitor {
263        type Value = Vec<AttemptEventRecord>;
264        fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
265            f.write_str("a bounded chronological attempt event list")
266        }
267        fn visit_seq<A: de::SeqAccess<'de>>(
268            self,
269            mut sequence: A,
270        ) -> Result<Self::Value, A::Error> {
271            let mut events = Vec::new();
272            while events.len() < MAX_ATTEMPT_EVENTS {
273                match sequence.next_element()? {
274                    Some(event) => events.push(event),
275                    None => return Ok(events),
276                }
277            }
278            if sequence.next_element::<de::IgnoredAny>()?.is_some() {
279                return Err(de::Error::custom(
280                    AttemptJournalRecordError::HistoryTooLarge,
281                ));
282            }
283            Ok(events)
284        }
285    }
286    deserializer.deserialize_seq(EventsVisitor)
287}
288
289/// Typed exact identity, finite accounting or chronological state rejection.
290#[derive(Debug, Error)]
291pub enum AttemptJournalRecordError {
292    /// Other product generations are not maintained.
293    #[error("unsupported attempt journal version {0}")]
294    UnsupportedVersion(u16),
295    /// Original limits overflow or exceed the bounded total.
296    #[error("attempt allowance exceeds {MAX_OPERATION_ATTEMPTS}")]
297    BudgetTooLarge,
298    /// The chronological metadata count exceeds its finite bound.
299    #[error("attempt history exceeds {MAX_ATTEMPT_EVENTS} events")]
300    HistoryTooLarge,
301    /// A selected principal is not admitted by its owning boundary.
302    #[error("invalid attempt operation principal")]
303    InvalidPrincipal,
304    /// A digest fails exact canonical admission.
305    #[error(transparent)]
306    Checksum(#[from] ChecksumError),
307    /// An unresolved paid mutation blocks another mutation.
308    #[error("mutation attempt {attempt} remains unresolved")]
309    MutationPending {
310        /// Retained unresolved attempt number.
311        attempt: u32,
312    },
313    /// An observation is still unsettled locally.
314    #[error("observation attempt {attempt} remains unresolved")]
315    ObservationPending {
316        /// Retained unresolved attempt number.
317        attempt: u32,
318    },
319    /// A receipt or reconciliation has no corresponding unresolved mutation.
320    #[error("no pending mutation attempt")]
321    NoPendingMutation,
322    /// A reply has no corresponding unresolved observation.
323    #[error("no pending observation attempt")]
324    NoPendingObservation,
325    /// A receipt or sequence targets another attempt.
326    #[error("attempt identity mismatch: expected {expected}, actual {actual}")]
327    AttemptMismatch {
328        /// Expected exact attempt number.
329        expected: u32,
330        /// Rejected attempt number.
331        actual: u32,
332    },
333    /// Reply/request identity differs from its retained exact reservation.
334    #[error("attempt request digest mismatch")]
335    RequestMismatch,
336    /// Original mutation authority has been consumed.
337    #[error("mutation attempt allowance exhausted")]
338    MutationBudgetExhausted,
339    /// Original observation authority has been consumed.
340    #[error("observation attempt allowance exhausted")]
341    ObservationBudgetExhausted,
342    /// This exact operation is already qualified as applied.
343    #[error("operation application already retained")]
344    AlreadyApplied,
345}
346
347#[cfg(test)]
348mod tests;
ic_backup/model/attempt_journal/mod.rs

ic_backup/model/attempt_journal/
mod.rs
