ic_backup/ops/persistence/download_journal/
mod.rs

1//! Locked durable local lifecycle updates and verified artifact publication.
2
3#[cfg(unix)]
4mod ic_snapshot_artifact;
5mod integrity;
6mod local_restore_source;
7mod manifest;
8#[cfg(unix)]
9mod metrics;
10#[cfg(unix)]
11pub use ic_snapshot_artifact::{
12    IcSnapshotArtifactError, IcSnapshotArtifactWriter, IcSnapshotDataUploadPreparationError,
13    IcSnapshotUploadArtifactError,
14};
15pub use integrity::DownloadIntegrityError;
16pub use local_restore_source::LocalRestoreSourceError;
17pub use manifest::{DownloadManifestError, read_download_manifest};
18#[cfg(unix)]
19pub use metrics::{
20    IcSnapshotLocalMetrics, MeasurementAggregate, MeasurementHistogram, MeasurementSummary,
21};
22
23use super::json::{check_json_size, create_encoded, encode_json, write_encoded};
24use super::{
25    BackupLayoutGuard, JournalLock, JournalLockError, PersistenceError, commit_artifact_directory,
26    read_json,
27};
28use crate::{
29    model::{
30        artifacts::{ArtifactChecksumRecord, canonical_hash},
31        download_journal::{
32            ArtifactStateRecord, DownloadArtifactRequest, DownloadJournalRecord,
33            DownloadJournalRecordError, MAX_DOWNLOAD_JOURNAL_BYTES,
34        },
35    },
36    ops::artifacts::{ArtifactError, checksum_directory},
37};
38use std::{fs, io, path::PathBuf};
39use thiserror::Error;
40
41const JOURNAL_FILE: &str = "download-journal.json";
42
43/// Exclusive local lifecycle access borrowing the stable backup layout guard.
44///
45/// The caller owns backend artifact completeness and fresh remote authority.
46/// These operations never invoke a transport, remove staging or release references.
47#[derive(Debug)]
48pub struct DownloadJournalGuard<'a> {
49    layout: &'a BackupLayoutGuard,
50    _lock: JournalLock,
51    record: DownloadJournalRecord,
52    usable: bool,
53    #[cfg(unix)]
54    ic_snapshot_metrics: std::sync::Mutex<IcSnapshotLocalMetrics>,
55}
56
57impl<'a> DownloadJournalGuard<'a> {
58    /// Exclusively create exact intent and snapshot identities without replacing evidence.
59    ///
60    /// # Errors
61    /// Rejects existing/unsafe journals, locked or replaced layouts and invalid/bounded records.
62    pub fn create(
63        layout: &'a BackupLayoutGuard,
64        intent: &str,
65        artifacts: Vec<DownloadArtifactRequest>,
66    ) -> Result<Self, DownloadJournalError> {
67        layout.check_root()?;
68        let path = layout.root().join(JOURNAL_FILE);
69        let lock = JournalLock::acquire(&path)?;
70        let record = DownloadJournalRecord::new(intent, artifacts)?;
71        let bytes = encode_json(&record, MAX_DOWNLOAD_JOURNAL_BYTES)?;
72        create_encoded(&path, &bytes)?;
73        Ok(Self {
74            layout,
75            _lock: lock,
76            record,
77            usable: true,
78            #[cfg(unix)]
79            ic_snapshot_metrics: std::sync::Mutex::default(),
80        })
81    }
82
83    /// Open retained bounded v1 evidence under exact caller-supplied intent.
84    ///
85    /// Reads only local journal evidence; it does not reverify artifacts or remote state.
86    /// # Errors
87    /// Rejects missing/unsafe/corrupt journals, intent mismatch and locked/replaced layouts.
88    pub fn open(
89        layout: &'a BackupLayoutGuard,
90        expected_intent: &str,
91    ) -> Result<Self, DownloadJournalError> {
92        layout.check_root()?;
93        let expected = canonical_hash(expected_intent).map_err(DownloadJournalRecordError::from)?;
94        let path = layout.root().join(JOURNAL_FILE);
95        let lock = JournalLock::acquire(&path)?;
96        let record: DownloadJournalRecord = read_json(&path, MAX_DOWNLOAD_JOURNAL_BYTES)?;
97        check_json_size(&record, MAX_DOWNLOAD_JOURNAL_BYTES)?;
98        if record.intent() != expected {
99            return Err(DownloadJournalError::IntentMismatch);
100        }
101        Ok(Self {
102            layout,
103            _lock: lock,
104            record,
105            usable: true,
106            #[cfg(unix)]
107            ic_snapshot_metrics: std::sync::Mutex::default(),
108        })
109    }
110
111    /// Read retained progress; failed publication requires reopening before further use.
112    ///
113    /// # Errors
114    /// Rejects an indeterminate write outcome or a replaced layout.
115    pub fn record(&self) -> Result<&DownloadJournalRecord, DownloadJournalError> {
116        self.check_usable()?;
117        Ok(&self.record)
118    }
119
120    /// Return the canonical journal location whose sidecar this guard owns.
121    #[must_use]
122    pub fn path(&self) -> PathBuf {
123        self.layout.root().join(JOURNAL_FILE)
124    }
125
126    /// Retain the caller's complete-download attestation for the exact snapshot.
127    ///
128    /// Requires a safe existing staging directory. The caller must already have
129    /// validated complete backend metadata/extent coverage and command quiescence;
130    /// traversability alone does not establish IC transfer completeness.
131    /// # Errors
132    /// Rejects identity/state conflicts, unsafe or missing staging and failed persistence.
133    pub fn record_downloaded(
134        &mut self,
135        canister: &str,
136        snapshot: &str,
137    ) -> Result<(), DownloadJournalError> {
138        let next = self.next(canister, snapshot, ArtifactStateRecord::Downloaded, None)?;
139        self.check_artifact_parent()?;
140        let entry = next.artifact(canister, snapshot)?;
141        checksum_directory(&self.layout.root().join(entry.staging_path()))?;
142        self.store(next, write_encoded)
143    }
144
145    /// Verify staged bytes and durably retain their canonical checksum.
146    ///
147    /// # Errors
148    /// Rejects wrong identity/state, unsafe or missing bytes and failed persistence.
149    pub fn verify_artifact(
150        &mut self,
151        canister: &str,
152        snapshot: &str,
153    ) -> Result<(), DownloadJournalError> {
154        self.check_usable()?;
155        let entry = self.record.artifact(canister, snapshot)?;
156        if entry.state() != ArtifactStateRecord::Downloaded {
157            return Err(DownloadJournalRecordError::InvalidStateTransition {
158                from: entry.state(),
159                to: ArtifactStateRecord::ChecksumVerified,
160            }
161            .into());
162        }
163        self.check_artifact_parent()?;
164        let checksum = checksum_directory(&self.layout.root().join(entry.staging_path()))?;
165        let next = self.next(
166            canister,
167            snapshot,
168            ArtifactStateRecord::ChecksumVerified,
169            Some(checksum),
170        )?;
171        self.store(next, write_encoded)
172    }
173
174    /// Publish exact verified bytes or adopt a matching tree after a lost response.
175    ///
176    /// Leaves staging and retained intent intact on rejection. Durable state does
177    /// not silently trigger fresh artifact verification; that is a distinct action.
178    /// # Errors
179    /// Rejects wrong identity/state, changed bytes, unsafe paths and uncertain publication.
180    pub fn finalize_artifact(
181        &mut self,
182        canister: &str,
183        snapshot: &str,
184    ) -> Result<(), DownloadJournalError> {
185        self.finalize_with(canister, snapshot, write_encoded)
186    }
187
188    fn finalize_with(
189        &mut self,
190        canister: &str,
191        snapshot: &str,
192        write: impl FnOnce(&std::path::Path, &[u8]) -> Result<(), PersistenceError>,
193    ) -> Result<(), DownloadJournalError> {
194        let next = self.next(canister, snapshot, ArtifactStateRecord::Durable, None)?;
195        let bytes = encode_json(&next, MAX_DOWNLOAD_JOURNAL_BYTES)?;
196        self.check_artifact_parent()?;
197        let entry = next.artifact(canister, snapshot)?;
198        let checksum = entry
199            .checksum()
200            .ok_or(DownloadJournalRecordError::InvalidChecksumState(
201                ArtifactStateRecord::Durable,
202            ))?;
203        // A failed commit may have published bytes. Stop this guard until its
204        // durable journal is reopened and those exact bytes are reconciled.
205        self.usable = false;
206        commit_artifact_directory(
207            &self.layout.root().join(entry.staging_path()),
208            &self.layout.root().join(entry.artifact_path()),
209            checksum.hash(),
210        )?;
211        self.store_encoded(next, &bytes, write)
212    }
213
214    fn next(
215        &self,
216        canister: &str,
217        snapshot: &str,
218        state: ArtifactStateRecord,
219        checksum: Option<ArtifactChecksumRecord>,
220    ) -> Result<DownloadJournalRecord, DownloadJournalError> {
221        self.check_usable()?;
222        let mut next = self.record.clone();
223        next.advance(canister, snapshot, state, checksum)?;
224        Ok(next)
225    }
226
227    fn check_usable(&self) -> Result<(), DownloadJournalError> {
228        if !self.usable {
229            return Err(DownloadJournalError::IndeterminateWrite);
230        }
231        self.layout.check_root()?;
232        Ok(())
233    }
234
235    fn check_artifact_parent(&self) -> Result<(), DownloadJournalError> {
236        let path = self.layout.root().join("artifacts");
237        if !fs::symlink_metadata(&path)?.is_dir() {
238            return Err(DownloadJournalError::UnsafeArtifactParent { path });
239        }
240        Ok(())
241    }
242
243    fn store(
244        &mut self,
245        next: DownloadJournalRecord,
246        write: impl FnOnce(&std::path::Path, &[u8]) -> Result<(), PersistenceError>,
247    ) -> Result<(), DownloadJournalError> {
248        let bytes = encode_json(&next, MAX_DOWNLOAD_JOURNAL_BYTES)?;
249        self.store_encoded(next, &bytes, write)
250    }
251
252    fn store_encoded(
253        &mut self,
254        next: DownloadJournalRecord,
255        bytes: &[u8],
256        write: impl FnOnce(&std::path::Path, &[u8]) -> Result<(), PersistenceError>,
257    ) -> Result<(), DownloadJournalError> {
258        self.layout.check_root()?;
259        self.usable = false;
260        write(&self.path(), bytes)?;
261        self.record = next;
262        self.usable = true;
263        Ok(())
264    }
265}
266
267/// Typed local journal admission or durable lifecycle failure.
268#[derive(Debug, Error)]
269pub enum DownloadJournalError {
270    /// The caller's exact intent digest differs from retained evidence.
271    #[error("download journal immutable intent mismatch")]
272    IntentMismatch,
273    /// A publication may have completed; reopen and reconcile retained evidence.
274    #[error("download journal outcome is indeterminate; reopen retained evidence")]
275    IndeterminateWrite,
276    /// The fixed artifact parent is not an existing regular directory.
277    #[error("unsafe download artifact parent: {path:?}")]
278    UnsafeArtifactParent {
279        /// Rejected location.
280        path: PathBuf,
281    },
282    /// Model identity, schema or transition admission failed.
283    #[error(transparent)]
284    Record(#[from] DownloadJournalRecordError),
285    /// Local journal or layout exclusion failed.
286    #[error(transparent)]
287    Lock(#[from] JournalLockError),
288    /// Durable local record or artifact publication failed.
289    #[error(transparent)]
290    Persistence(#[from] PersistenceError),
291    /// Secure artifact traversal failed.
292    #[error(transparent)]
293    Artifact(#[from] ArtifactError),
294    /// Local fixture-independent filesystem access failed.
295    #[error(transparent)]
296    Io(#[from] io::Error),
297}
298
299#[cfg(all(test, unix))]
300mod tests;
ic_backup/ops/persistence/download_journal/mod.rs

ic_backup/ops/persistence/download_journal/
mod.rs
