ic_backup/model/attempt_journal/
mod.rs

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

ic_backup/model/attempt_journal/
mod.rs
