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

ic_backup/model/attempt_journal/
mod.rs
