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/// Exact locally retained evidence for an Applied operation.
97///
98/// The integration owns the evidence's meaning and qualification. In particular,
99/// observation evidence is not the original mutation reply or retry authority.
100#[derive(Clone, Copy, Debug, Eq, PartialEq)]
101pub enum AppliedReceiptEvidence<'a> {
102    /// The original mutation was settled by its direct qualified receipt.
103    Mutation {
104        /// Original mutation attempt number.
105        attempt: u32,
106        /// Digest of the retained integration-qualified direct evidence.
107        evidence: &'a str,
108    },
109    /// A separately budgeted observation settled the original mutation.
110    Observation {
111        /// Separately reserved observation attempt number.
112        attempt: u32,
113        /// Exact observation request digest from its reservation.
114        request: &'a str,
115        /// Digest of retained integration-qualified observation evidence.
116        evidence: &'a str,
117    },
118}
119
120/// Maintained v1 append-only local reservations and receipts for one exact operation.
121#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
122#[serde(try_from = "JournalFields")]
123pub struct AttemptJournalRecord {
124    version: u16,
125    authority: AttemptAuthorityRecord,
126    events: Vec<AttemptEventRecord>,
127    #[serde(skip)]
128    projection: Projection,
129}
130
131#[derive(Deserialize)]
132#[serde(deny_unknown_fields)]
133struct JournalFields {
134    version: u16,
135    authority: AttemptAuthorityRecord,
136    #[serde(deserialize_with = "bounded_events")]
137    events: Vec<AttemptEventRecord>,
138}
139
140impl TryFrom<JournalFields> for AttemptJournalRecord {
141    type Error = AttemptJournalRecordError;
142    fn try_from(mut fields: JournalFields) -> Result<Self, Self::Error> {
143        if fields.version != 1 {
144            return Err(AttemptJournalRecordError::UnsupportedVersion(
145                fields.version,
146            ));
147        }
148        let mut projection = Projection::empty();
149        for event in &mut fields.events {
150            event.normalize()?;
151            projection.apply(event, fields.authority.budget())?;
152        }
153        Ok(Self {
154            version: 1,
155            authority: fields.authority,
156            events: fields.events,
157            projection,
158        })
159    }
160}
161
162impl AttemptJournalRecord {
163    /// Retain an exact validated authority with no consumed attempts.
164    #[must_use]
165    pub const fn new(authority: AttemptAuthorityRecord) -> Self {
166        Self {
167            version: 1,
168            authority,
169            events: Vec::new(),
170            projection: Projection::empty(),
171        }
172    }
173    /// Read exact declared identity and immutable ceilings.
174    #[must_use]
175    pub const fn authority(&self) -> &AttemptAuthorityRecord {
176        &self.authority
177    }
178    /// Read the exact canonical request of the unresolved observation, if any.
179    ///
180    /// This projects existing validated history. It creates no reservation or
181    /// repeat-dispatch permission and returns None after a settled observation.
182    #[must_use]
183    pub fn pending_observation_request(&self) -> Option<&str> {
184        self.projection
185            .pending_observation
186            .as_ref()
187            .map(|pending| pending.request.as_str())
188    }
189    /// Read the terminal Applied receipt from validated original history.
190    ///
191    /// This does not authenticate evidence or grant dispatch, retry or release.
192    #[must_use]
193    pub fn applied_receipt(&self) -> Option<AppliedReceiptEvidence<'_>> {
194        match self.events.last()? {
195            AttemptEventRecord::MutationResolved {
196                mutation,
197                outcome: MutationOutcomeRecord::Applied,
198                evidence,
199            } => Some(AppliedReceiptEvidence::Mutation {
200                attempt: *mutation,
201                evidence,
202            }),
203            AttemptEventRecord::ObservationRecorded {
204                observation,
205                request,
206                outcome: ObservationOutcomeRecord::Applied,
207                evidence,
208            } => Some(AppliedReceiptEvidence::Observation {
209                attempt: *observation,
210                request,
211                evidence,
212            }),
213            _ => None,
214        }
215    }
216    /// Project local evidence without IO or replenishing authority.
217    ///
218    /// Use `Self::applied_receipt` for the terminal receipt or `Self::digest`
219    /// for exact full-history identity.
220    #[must_use]
221    pub fn view(&self) -> AttemptJournalView {
222        AttemptJournalView {
223            authority: self.authority.digest(),
224            mutations_used: self.projection.mutations_used,
225            observations_used: self.projection.observations_used,
226            mutations_remaining: self.authority.budget().mutations()
227                - self.projection.mutations_used,
228            observations_remaining: self.authority.budget().observations()
229                - self.projection.observations_used,
230            pending_mutation: self.projection.pending_mutation,
231            pending_observation: self
232                .projection
233                .pending_observation
234                .as_ref()
235                .map(|pending| pending.attempt),
236            applied: self.projection.applied,
237            history_len: self.events.len(),
238        }
239    }
240    /// Hash full original authority and every canonical chronological reservation/receipt.
241    ///
242    /// The NUL-terminated v1 domain precedes 64 ASCII authority hash bytes, a
243    /// big-endian u64 event count and closed tagged event bytes. Derived projections,
244    /// JSON formatting and filesystem paths are excluded. This authenticates no receipt.
245    #[must_use]
246    pub fn digest(&self) -> ArtifactChecksumRecord {
247        let mut bytes = b"ic-backup/attempt-journal-history/v1\0".to_vec();
248        bytes.extend_from_slice(self.authority.digest().hash().as_bytes());
249        bytes.extend_from_slice(&(self.events.len() as u64).to_be_bytes());
250        for event in &self.events {
251            event.append_digest_bytes(&mut bytes);
252        }
253        ArtifactChecksumRecord::from_bytes(&bytes)
254    }
255    pub(crate) fn reserve_mutation(&mut self) -> Result<u32, AttemptJournalRecordError> {
256        let attempt = self.projection.next_attempt;
257        self.append(AttemptEventRecord::MutationReserved { attempt })?;
258        Ok(attempt)
259    }
260    pub(crate) fn reserve_observation(
261        &mut self,
262        mutation: u32,
263        request: &str,
264    ) -> Result<u32, AttemptJournalRecordError> {
265        let attempt = self.projection.next_attempt;
266        self.append(AttemptEventRecord::ObservationReserved {
267            attempt,
268            mutation,
269            request: request.to_owned(),
270        })?;
271        Ok(attempt)
272    }
273    pub(crate) fn record_mutation(
274        &mut self,
275        receipt: MutationReceiptRequest,
276    ) -> Result<(), AttemptJournalRecordError> {
277        if canonical_hash(&receipt.request)? != self.authority.binding().request() {
278            return Err(AttemptJournalRecordError::RequestMismatch);
279        }
280        self.append(AttemptEventRecord::MutationResolved {
281            mutation: receipt.attempt,
282            outcome: receipt.outcome,
283            evidence: receipt.evidence,
284        })
285    }
286    pub(crate) fn record_observation(
287        &mut self,
288        receipt: ObservationReceiptRequest,
289    ) -> Result<(), AttemptJournalRecordError> {
290        self.append(AttemptEventRecord::ObservationRecorded {
291            observation: receipt.attempt,
292            request: receipt.request,
293            outcome: receipt.outcome,
294            evidence: receipt.evidence,
295        })
296    }
297    fn append(&mut self, mut event: AttemptEventRecord) -> Result<(), AttemptJournalRecordError> {
298        if self.events.len() == MAX_ATTEMPT_EVENTS {
299            return Err(AttemptJournalRecordError::HistoryTooLarge);
300        }
301        event.normalize()?;
302        let mut projection = self.projection.clone();
303        projection.apply(&event, self.authority.budget())?;
304        self.events.push(event);
305        self.projection = projection;
306        Ok(())
307    }
308}
309
310fn bounded_events<'de, D: Deserializer<'de>>(
311    deserializer: D,
312) -> Result<Vec<AttemptEventRecord>, D::Error> {
313    struct EventsVisitor;
314    impl<'de> de::Visitor<'de> for EventsVisitor {
315        type Value = Vec<AttemptEventRecord>;
316        fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
317            f.write_str("a bounded chronological attempt event list")
318        }
319        fn visit_seq<A: de::SeqAccess<'de>>(
320            self,
321            mut sequence: A,
322        ) -> Result<Self::Value, A::Error> {
323            let mut events = Vec::new();
324            while events.len() < MAX_ATTEMPT_EVENTS {
325                match sequence.next_element()? {
326                    Some(event) => events.push(event),
327                    None => return Ok(events),
328                }
329            }
330            if sequence.next_element::<de::IgnoredAny>()?.is_some() {
331                return Err(de::Error::custom(
332                    AttemptJournalRecordError::HistoryTooLarge,
333                ));
334            }
335            Ok(events)
336        }
337    }
338    deserializer.deserialize_seq(EventsVisitor)
339}
340
341/// Typed exact identity, finite accounting or chronological state rejection.
342#[derive(Debug, Error)]
343pub enum AttemptJournalRecordError {
344    /// Other product generations are not maintained.
345    #[error("unsupported attempt journal version {0}")]
346    UnsupportedVersion(u16),
347    /// Original limits overflow or exceed the bounded total.
348    #[error("attempt allowance exceeds {MAX_OPERATION_ATTEMPTS}")]
349    BudgetTooLarge,
350    /// The chronological metadata count exceeds its finite bound.
351    #[error("attempt history exceeds {MAX_ATTEMPT_EVENTS} events")]
352    HistoryTooLarge,
353    /// A selected principal is not admitted by its owning boundary.
354    #[error("invalid attempt operation principal")]
355    InvalidPrincipal,
356    /// A digest fails exact canonical admission.
357    #[error(transparent)]
358    Checksum(#[from] ChecksumError),
359    /// An unresolved paid mutation blocks another mutation.
360    #[error("mutation attempt {attempt} remains unresolved")]
361    MutationPending {
362        /// Retained unresolved attempt number.
363        attempt: u32,
364    },
365    /// An observation is still unsettled locally.
366    #[error("observation attempt {attempt} remains unresolved")]
367    ObservationPending {
368        /// Retained unresolved attempt number.
369        attempt: u32,
370    },
371    /// A receipt or reconciliation has no corresponding unresolved mutation.
372    #[error("no pending mutation attempt")]
373    NoPendingMutation,
374    /// A reply has no corresponding unresolved observation.
375    #[error("no pending observation attempt")]
376    NoPendingObservation,
377    /// A receipt or sequence targets another attempt.
378    #[error("attempt identity mismatch: expected {expected}, actual {actual}")]
379    AttemptMismatch {
380        /// Expected exact attempt number.
381        expected: u32,
382        /// Rejected attempt number.
383        actual: u32,
384    },
385    /// Reply/request identity differs from its retained exact reservation.
386    #[error("attempt request digest mismatch")]
387    RequestMismatch,
388    /// Original mutation authority has been consumed.
389    #[error("mutation attempt allowance exhausted")]
390    MutationBudgetExhausted,
391    /// Original observation authority has been consumed.
392    #[error("observation attempt allowance exhausted")]
393    ObservationBudgetExhausted,
394    /// This exact operation is already qualified as applied.
395    #[error("operation application already retained")]
396    AlreadyApplied,
397}
398
399#[cfg(test)]
400mod tests;
ic_backup/model/attempt_journal/mod.rs

ic_backup/model/attempt_journal/
mod.rs
