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;
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<E>(
188        &self,
189        authority: &AttemptAuthorityRecord,
190        mutation_attempt: u32,
191        authority_mismatch: E,
192        mutation_mismatch: E,
193        observation_pending: E,
194    ) -> Result<(), E> {
195        if self.authority() != authority {
196            return Err(authority_mismatch);
197        }
198        if self.pending_mutation() != Some(mutation_attempt) {
199            return Err(mutation_mismatch);
200        }
201        if self.pending_observation().is_some() {
202            return Err(observation_pending);
203        }
204        Ok(())
205    }
206    // Reservation admission reads the validated projection without building the
207    // public reporting view or hashing authority unrelated to these attempt IDs.
208    pub(crate) const fn pending_mutation(&self) -> Option<u32> {
209        self.projection.pending_mutation()
210    }
211    pub(crate) const fn pending_observation(&self) -> Option<u32> {
212        match self.projection.pending_observation() {
213            Some(pending) => Some(pending.attempt),
214            None => None,
215        }
216    }
217    /// Read the exact canonical request of the unresolved observation, if any.
218    ///
219    /// This projects existing validated history. It creates no reservation or
220    /// repeat-dispatch permission and returns None after a settled observation.
221    #[must_use]
222    pub fn pending_observation_request(&self) -> Option<&str> {
223        self.projection
224            .pending_observation()
225            .map(|pending| pending.request.as_str())
226    }
227    /// Read the terminal Applied receipt from validated original history.
228    ///
229    /// This does not authenticate evidence or grant dispatch, retry or release.
230    #[must_use]
231    pub fn applied_receipt(&self) -> Option<AppliedReceiptEvidence<'_>> {
232        match self.events.last()? {
233            AttemptEventRecord::MutationResolved {
234                mutation,
235                outcome: MutationOutcomeRecord::Applied,
236                evidence,
237            } => Some(AppliedReceiptEvidence::Mutation {
238                attempt: *mutation,
239                evidence,
240            }),
241            AttemptEventRecord::ObservationRecorded {
242                observation,
243                request,
244                outcome: ObservationOutcomeRecord::Applied,
245                evidence,
246            } => Some(AppliedReceiptEvidence::Observation {
247                attempt: *observation,
248                request,
249                evidence,
250            }),
251            _ => None,
252        }
253    }
254    /// Project local evidence without IO or replenishing authority.
255    ///
256    /// Use `Self::applied_receipt` for the terminal receipt or `Self::digest`
257    /// for exact full-history identity.
258    #[must_use]
259    pub fn view(&self) -> AttemptJournalView {
260        AttemptJournalView {
261            authority: self.authority.digest(),
262            mutations_used: self.projection.mutations_used,
263            observations_used: self.projection.observations_used,
264            mutations_remaining: self.authority.budget().mutations()
265                - self.projection.mutations_used,
266            observations_remaining: self.authority.budget().observations()
267                - self.projection.observations_used,
268            pending_mutation: self.projection.pending_mutation(),
269            pending_observation: self
270                .projection
271                .pending_observation()
272                .map(|pending| pending.attempt),
273            applied: self.projection.applied(),
274            history_len: self.events.len(),
275        }
276    }
277    /// Hash full original authority and every canonical chronological reservation/receipt.
278    ///
279    /// The NUL-terminated v1 domain precedes 64 ASCII authority hash bytes, a
280    /// big-endian u64 event count and closed tagged event bytes. Derived projections,
281    /// JSON formatting and filesystem paths are excluded. This authenticates no receipt.
282    #[must_use]
283    pub fn digest(&self) -> ArtifactChecksumRecord {
284        let mut bytes = EvidenceDigest::new(b"ic-backup/attempt-journal-history/v1\0");
285        bytes.checksum(&self.authority.digest());
286        bytes.u64_be(self.events.len() as u64);
287        for event in &self.events {
288            event.append_digest_bytes(&mut bytes);
289        }
290        bytes.finish()
291    }
292    pub(crate) fn reserve_mutation(&mut self) -> Result<u32, AttemptJournalRecordError> {
293        let attempt = self.projection.next_attempt;
294        self.append(AttemptEventRecord::MutationReserved { attempt })?;
295        Ok(attempt)
296    }
297    pub(crate) fn reserve_observation(
298        &mut self,
299        mutation: u32,
300        request: &str,
301    ) -> Result<u32, AttemptJournalRecordError> {
302        let attempt = self.projection.next_attempt;
303        self.append(AttemptEventRecord::ObservationReserved {
304            attempt,
305            mutation,
306            request: request.to_owned(),
307        })?;
308        Ok(attempt)
309    }
310    pub(crate) fn record_mutation(
311        &mut self,
312        receipt: MutationReceiptRequest,
313    ) -> Result<(), AttemptJournalRecordError> {
314        if canonical_hash(&receipt.request)? != self.authority.binding().request() {
315            return Err(AttemptJournalRecordError::RequestMismatch);
316        }
317        self.append(AttemptEventRecord::MutationResolved {
318            mutation: receipt.attempt,
319            outcome: receipt.outcome,
320            evidence: receipt.evidence,
321        })
322    }
323    pub(crate) fn record_observation(
324        &mut self,
325        receipt: ObservationReceiptRequest,
326    ) -> Result<(), AttemptJournalRecordError> {
327        self.append(AttemptEventRecord::ObservationRecorded {
328            observation: receipt.attempt,
329            request: receipt.request,
330            outcome: receipt.outcome,
331            evidence: receipt.evidence,
332        })
333    }
334    fn append(&mut self, mut event: AttemptEventRecord) -> Result<(), AttemptJournalRecordError> {
335        if self.events.len() == MAX_ATTEMPT_EVENTS {
336            return Err(AttemptJournalRecordError::HistoryTooLarge);
337        }
338        event.normalize()?;
339        let mut projection = self.projection.clone();
340        projection.apply(&event, self.authority.budget())?;
341        self.events.push(event);
342        self.projection = projection;
343        Ok(())
344    }
345}
346
347fn bounded_events<'de, D: Deserializer<'de>>(
348    deserializer: D,
349) -> Result<Vec<AttemptEventRecord>, D::Error> {
350    bounded_sequence::deserialize::<_, _, MAX_ATTEMPT_EVENTS>(
351        deserializer,
352        "a bounded chronological attempt event list",
353        AttemptJournalRecordError::HistoryTooLarge,
354    )
355}
356
357/// Typed exact identity, finite accounting or chronological state rejection.
358#[derive(Debug, Error)]
359pub enum AttemptJournalRecordError {
360    /// Other product generations are not maintained.
361    #[error("unsupported attempt journal version {0}")]
362    UnsupportedVersion(u16),
363    /// Original limits overflow or exceed the bounded total.
364    #[error("attempt allowance exceeds {MAX_OPERATION_ATTEMPTS}")]
365    BudgetTooLarge,
366    /// The chronological metadata count exceeds its finite bound.
367    #[error("attempt history exceeds {MAX_ATTEMPT_EVENTS} events")]
368    HistoryTooLarge,
369    /// A selected principal is not admitted by its owning boundary.
370    #[error("invalid attempt operation principal")]
371    InvalidPrincipal,
372    /// A digest fails exact canonical admission.
373    #[error(transparent)]
374    Checksum(#[from] ChecksumError),
375    /// An unresolved paid mutation blocks another mutation.
376    #[error("mutation attempt {attempt} remains unresolved")]
377    MutationPending {
378        /// Retained unresolved attempt number.
379        attempt: u32,
380    },
381    /// An observation is still unsettled locally.
382    #[error("observation attempt {attempt} remains unresolved")]
383    ObservationPending {
384        /// Retained unresolved attempt number.
385        attempt: u32,
386    },
387    /// A receipt or reconciliation has no corresponding unresolved mutation.
388    #[error("no pending mutation attempt")]
389    NoPendingMutation,
390    /// A reply has no corresponding unresolved observation.
391    #[error("no pending observation attempt")]
392    NoPendingObservation,
393    /// A receipt or sequence targets another attempt.
394    #[error("attempt identity mismatch: expected {expected}, actual {actual}")]
395    AttemptMismatch {
396        /// Expected exact attempt number.
397        expected: u32,
398        /// Rejected attempt number.
399        actual: u32,
400    },
401    /// Reply/request identity differs from its retained exact reservation.
402    #[error("attempt request digest mismatch")]
403    RequestMismatch,
404    /// Original mutation authority has been consumed.
405    #[error("mutation attempt allowance exhausted")]
406    MutationBudgetExhausted,
407    /// Original observation authority has been consumed.
408    #[error("observation attempt allowance exhausted")]
409    ObservationBudgetExhausted,
410    /// This exact operation is already qualified as applied.
411    #[error("operation application already retained")]
412    AlreadyApplied,
413}
414
415#[cfg(test)]
416mod tests;
ic_backup/model/attempt_journal/mod.rs

ic_backup/model/attempt_journal/
mod.rs
