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

ic_backup/model/attempt_journal/
mod.rs
