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

ic_backup/ops/persistence/attempt_journal/
mod.rs
