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

ic_backup/ops/persistence/attempt_journal/
mod.rs
