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

ic_backup/model/attempt_journal/
mod.rs
