ic_backup/ops/persistence/attempt_journal/
mod.rs

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

ic_backup/ops/persistence/attempt_journal/
mod.rs
