ic_backup/ops/persistence/attempt_journal/
mod.rs

1//! Exclusive durable attempt reservations and exact receipt retention; no transport.
2
3use super::json::{create_encoded, encode_json, read_record, write_encoded};
4use super::{
5    BackupLayoutGuard, ExecutionProgressPersistenceError, JournalLock, JournalLockError,
6    OperationPlanPersistenceError, PersistenceError, read_operation_plan,
7};
8use crate::model::artifacts::ArtifactChecksumRecord;
9use crate::model::attempt_journal::{
10    AttemptAuthorityRecord, AttemptJournalRecord, AttemptJournalRecordError,
11    MAX_ATTEMPT_JOURNAL_BYTES, MutationReceiptRequest, ObservationReceiptRequest,
12};
13use crate::model::operation_plan::{OperationPlanError, OperationPlanRecord};
14use std::path::{Path, PathBuf};
15use thiserror::Error;
16
17/// Exclusive local accounting borrowing stable layout exclusion.
18///
19/// Successful reservation persists consumption before returning its number.
20/// It grants no fresh IC authority, backend settlement or subprocess dispatch permission.
21#[derive(Debug)]
22pub struct AttemptJournalGuard<'a> {
23    layout: &'a BackupLayoutGuard,
24    _lock: JournalLock,
25    record: AttemptJournalRecord,
26    usable: bool,
27}
28
29impl<'a> AttemptJournalGuard<'a> {
30    /// Create original exact identity and limits without replacing retained evidence.
31    ///
32    /// # Errors
33    /// Rejects unsafe/existing journals, lock contention, replaced layouts and failed persistence.
34    pub fn create(
35        layout: &'a BackupLayoutGuard,
36        authority: AttemptAuthorityRecord,
37    ) -> Result<Self, AttemptJournalError> {
38        layout.check_root()?;
39        let path = journal_path(layout, &authority);
40        let lock = JournalLock::acquire(&path)?;
41        let record = AttemptJournalRecord::new(authority);
42        let bytes = encode_json(&record, MAX_ATTEMPT_JOURNAL_BYTES)?;
43        create_encoded(&path, &bytes)?;
44        Ok(Self {
45            layout,
46            _lock: lock,
47            record,
48            usable: true,
49        })
50    }
51    /// Open exact retained identity and original budgets, replaying only local events.
52    ///
53    /// # Errors
54    /// Rejects identity/budget changes, invalid chronology, missing/unsafe/oversized records and locks.
55    pub fn open(
56        layout: &'a BackupLayoutGuard,
57        expected: &AttemptAuthorityRecord,
58    ) -> Result<Self, AttemptJournalError> {
59        layout.check_root()?;
60        let path = journal_path(layout, expected);
61        let lock = JournalLock::acquire(&path)?;
62        let record: AttemptJournalRecord = read_record(&path, MAX_ATTEMPT_JOURNAL_BYTES)?;
63        if record.authority() != expected {
64            return Err(AttemptJournalError::AuthorityMismatch);
65        }
66        Ok(Self {
67            layout,
68            _lock: lock,
69            record,
70            usable: true,
71        })
72    }
73    /// Read validated retained accounting without observing remote state.
74    ///
75    /// # Errors
76    /// Rejects indeterminate writes or replaced layouts.
77    pub fn record(&self) -> Result<&AttemptJournalRecord, AttemptJournalError> {
78        self.check_usable()?;
79        Ok(&self.record)
80    }
81
82    fn into_record(self) -> Result<AttemptJournalRecord, AttemptJournalError> {
83        self.check_usable()?;
84        Ok(self.record)
85    }
86    /// Return the exact journal location for this operation sequence.
87    #[must_use]
88    pub fn path(&self) -> PathBuf {
89        journal_path(self.layout, self.record.authority())
90    }
91    /// Durably consume one mutation attempt before any admitted caller-owned call.
92    ///
93    /// # Errors
94    /// Rejects unresolved attempts, exhaustion, applied operations and persistence failures.
95    pub(crate) fn reserve_mutation(&mut self) -> Result<u32, AttemptJournalError> {
96        self.reserve_with(AttemptJournalRecord::reserve_mutation, write_encoded)
97    }
98    /// Reserve only after admitting the exact persisted plan and all original journals.
99    ///
100    /// Every declared prerequisite must retain Applied evidence. Hold no other attempt
101    /// guard; original journals are admitted with sequential locks while this guard
102    /// retains its own lock. No missing journal is created or treated as unused.
103    /// Fresh permissions, authenticated prerequisites, never-dispatched custody and
104    /// backend/application safety remain caller-owned. Success is accounting only.
105    /// # Errors
106    /// Rejects wrong/missing originals, contention, unmet dependencies and ordinary
107    /// pending/exhausted/Applied or indeterminate-write conditions without refunds.
108    pub fn reserve_planned_mutation(
109        &mut self,
110        expected_plan: &ArtifactChecksumRecord,
111    ) -> Result<u32, ExecutionProgressPersistenceError> {
112        super::execution_progress::admit_with_current(
113            self.layout,
114            expected_plan,
115            self.record()?,
116            true,
117        )?;
118        Ok(self.reserve_mutation()?)
119    }
120    /// Durably consume one observation bound to an exact unresolved mutation and request.
121    ///
122    /// Fresh authority, paid-effect settlement and ended command custody remain caller-owned.
123    /// # Errors
124    /// Rejects wrong attempts/requests, unresolved observations, exhaustion and failed writes.
125    pub(crate) fn reserve_observation(
126        &mut self,
127        mutation: u32,
128        request: &str,
129    ) -> Result<u32, AttemptJournalError> {
130        self.reserve_with(
131            |record| record.reserve_observation(mutation, request),
132            write_encoded,
133        )
134    }
135    /// Reserve one original reconciliation observation after complete retained-plan admission.
136    ///
137    /// Reuses the same mutation/observation accounting owner; complete original evidence
138    /// is required even after exhaustion or reopening. Hold no other attempt guard.
139    /// Fresh read permission, observation dispatch custody and settled authenticated
140    /// outcome qualification remain integration-owned. Lost replies stay pending.
141    /// # Errors
142    /// Rejects missing/changed/held originals, invalid causality and the existing
143    /// reservation/persistence failures. No journal or allowance is recreated.
144    pub fn reserve_planned_observation(
145        &mut self,
146        expected_plan: &ArtifactChecksumRecord,
147        mutation: u32,
148        request: &str,
149    ) -> Result<u32, ExecutionProgressPersistenceError> {
150        self.admit_planned_progress(expected_plan)?;
151        Ok(self.reserve_observation(mutation, request)?)
152    }
153    /// Retain an exact integration-qualified direct mutation reply.
154    ///
155    /// # Errors
156    /// Rejects unmatched identity, absent/unsettled attempts, invalid evidence and failed persistence.
157    pub fn record_mutation(
158        &mut self,
159        receipt: MutationReceiptRequest,
160    ) -> Result<(), AttemptJournalError> {
161        self.reserve_with(|record| record.record_mutation(receipt), write_encoded)
162    }
163    /// Retain an exact observation; unresolved evidence never authorizes another mutation.
164    ///
165    /// The integration qualifies completed observation and paid-effect settlement.
166    /// A lost observation response stays pending until qualified settlement.
167    ///
168    /// # Errors
169    /// Rejects unmatched identity, absent/unsettled attempts, invalid evidence and failed persistence.
170    pub fn record_observation(
171        &mut self,
172        receipt: ObservationReceiptRequest,
173    ) -> Result<(), AttemptJournalError> {
174        self.reserve_with(|record| record.record_observation(receipt), write_encoded)
175    }
176
177    /// Record an independently qualified observation after complete original-plan admission.
178    ///
179    /// Hold no other attempt guards. Original journals are read with sequential locks;
180    /// missing evidence never means unused allowance. This reuses the existing receipt
181    /// transition and durable writer, without another reservation or spending owner.
182    /// Authentication, settled outcome and retained evidence remain integration-owned.
183    /// # Errors
184    /// Rejects missing/changed/held originals, invalid causality or receipt, and failed IO.
185    pub fn record_planned_observation(
186        &mut self,
187        expected_plan: &ArtifactChecksumRecord,
188        receipt: ObservationReceiptRequest,
189    ) -> Result<(), ExecutionProgressPersistenceError> {
190        self.admit_planned_progress(expected_plan)?;
191        Ok(self.record_observation(receipt)?)
192    }
193
194    pub(crate) fn admit_planned_progress(
195        &self,
196        expected_plan: &ArtifactChecksumRecord,
197    ) -> Result<(), ExecutionProgressPersistenceError> {
198        super::execution_progress::admit_with_current(
199            self.layout,
200            expected_plan,
201            self.record()?,
202            false,
203        )?;
204        Ok(())
205    }
206
207    fn reserve_with<T>(
208        &mut self,
209        transition: impl FnOnce(&mut AttemptJournalRecord) -> Result<T, AttemptJournalRecordError>,
210        write: impl FnOnce(&Path, &[u8]) -> Result<(), PersistenceError>,
211    ) -> Result<T, AttemptJournalError> {
212        self.check_usable()?;
213        let mut next = self.record.clone();
214        let result = transition(&mut next)?;
215        let bytes = encode_json(&next, MAX_ATTEMPT_JOURNAL_BYTES)?;
216        let path = self.path();
217        super::json::replace_cached_record(&mut self.record, &mut self.usable, next, || {
218            write(&path, &bytes)
219        })?;
220        Ok(result)
221    }
222    fn check_usable(&self) -> Result<(), AttemptJournalError> {
223        if !self.usable {
224            return Err(AttemptJournalError::IndeterminateWrite);
225        }
226        self.layout.check_root()?;
227        Ok(())
228    }
229}
230
231pub(super) fn read_original_journals(
232    layout: &BackupLayoutGuard,
233    authorities: &[AttemptAuthorityRecord],
234    current: Option<&AttemptJournalRecord>,
235) -> Result<Vec<AttemptJournalRecord>, AttemptJournalError> {
236    layout.check_root()?;
237    if let Some(current) = current
238        && !authorities.contains(current.authority())
239    {
240        return Err(AttemptJournalError::AuthorityMismatch);
241    }
242    let mut journals = Vec::with_capacity(authorities.len());
243    for authority in authorities {
244        if let Some(current) = current
245            && current.authority() == authority
246        {
247            journals.push(current.clone());
248        } else {
249            let guard = AttemptJournalGuard::open(layout, authority)?;
250            journals.push(guard.into_record()?);
251        }
252    }
253    layout.check_root()?;
254    Ok(journals)
255}
256
257pub(super) fn read_original_plan_journals<E>(
258    layout: &BackupLayoutGuard,
259    expected_plan: &ArtifactChecksumRecord,
260    current: Option<&AttemptJournalRecord>,
261) -> Result<(OperationPlanRecord, Vec<AttemptJournalRecord>), E>
262where
263    E: From<OperationPlanPersistenceError> + From<OperationPlanError> + From<AttemptJournalError>,
264{
265    let plan = read_operation_plan(layout, expected_plan)?;
266    let authorities = plan.attempt_authorities()?;
267    let journals = read_original_journals(layout, &authorities, current)?;
268    Ok((plan, journals))
269}
270
271fn journal_path(layout: &BackupLayoutGuard, authority: &AttemptAuthorityRecord) -> PathBuf {
272    layout.root().join(format!(
273        "attempt-{}.json",
274        authority.binding().operation_sequence()
275    ))
276}
277
278/// Typed exact admission, local accounting or durable publication failure.
279#[derive(Debug, Error)]
280pub enum AttemptJournalError {
281    /// Retained identity or original ceilings differ from current exact selection.
282    #[error("attempt journal authority mismatch")]
283    AuthorityMismatch,
284    /// A write may have completed; drop and reopen exact retained evidence.
285    #[error("attempt journal write outcome indeterminate; reopen retained evidence")]
286    IndeterminateWrite,
287    /// Model identity, chronology or allowance admission failed.
288    #[error(transparent)]
289    Record(#[from] AttemptJournalRecordError),
290    /// Journal/layout exclusion failed.
291    #[error(transparent)]
292    Lock(#[from] JournalLockError),
293    /// Bounded durable record access failed.
294    #[error(transparent)]
295    Persistence(#[from] PersistenceError),
296}
297
298#[cfg(all(test, unix))]
299mod tests;
ic_backup/ops/persistence/attempt_journal/mod.rs

ic_backup/ops/persistence/attempt_journal/
mod.rs
