appcore_update/
store.rs

1use crate::filesystem::read_regular_file_bounded;
2use crate::{sha256_hex, ArtifactDescriptor, UpdateError, UpdateResult};
3use appcore_contracts::BuildId;
4use serde::{Deserialize, Serialize};
5#[cfg(unix)]
6use std::fs::File;
7use std::fs::{self, OpenOptions};
8use std::io::Write;
9use std::path::{Path, PathBuf};
10use std::sync::atomic::{AtomicU64, Ordering};
11
12/// Stable format version for update pointers and pending activation metadata.
13pub const UPDATE_METADATA_FORMAT_VERSION: u16 = 1;
14static UPDATE_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
15
16/// Opaque staged artifact owned by an artifact store.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct StagedArtifact {
19    /// Descriptor staged for activation.
20    pub descriptor: ArtifactDescriptor,
21    /// Store-owned staging reference.
22    pub staging_reference: String,
23}
24
25/// Receipt required to commit or roll back one activation.
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27pub struct ActivationReceipt {
28    /// Artifact that was activated.
29    pub activated: ArtifactDescriptor,
30    /// Previously active artifact, when one existed.
31    pub previous: Option<ArtifactDescriptor>,
32}
33
34/// Store contract for atomic staging and reversible activation.
35pub trait ArtifactStore: Send + Sync {
36    /// Recovers an activation interrupted before commit.
37    ///
38    /// Stores without durable activation metadata may keep the default no-op.
39    fn recover(&self) -> UpdateResult<()> {
40        Ok(())
41    }
42    /// Returns the currently active artifact.
43    fn current(&self) -> UpdateResult<Option<ArtifactDescriptor>>;
44    /// Persists verified bytes without changing the active artifact.
45    fn stage(&self, descriptor: &ArtifactDescriptor, bytes: &[u8]) -> UpdateResult<StagedArtifact>;
46    /// Removes a staged artifact that failed a pre-activation smoke test.
47    fn discard_staged(&self, _staged: &StagedArtifact) -> UpdateResult<()> {
48        Ok(())
49    }
50    /// Atomically makes a staged artifact active and returns rollback state.
51    fn activate(&self, staged: StagedArtifact) -> UpdateResult<ActivationReceipt>;
52    /// Restores the previous artifact from an activation receipt.
53    fn rollback(&self, receipt: &ActivationReceipt) -> UpdateResult<()>;
54    /// Finalizes a healthy activation and discards rollback metadata.
55    fn commit(&self, receipt: &ActivationReceipt) -> UpdateResult<()>;
56}
57
58/// Filesystem artifact store using atomic pointer replacement.
59#[derive(Debug, Clone)]
60pub struct FileArtifactStore {
61    root: PathBuf,
62}
63
64#[derive(Debug, Clone, Serialize, Deserialize)]
65struct ArtifactPointer {
66    format_version: u16,
67    descriptor: ArtifactDescriptor,
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize)]
71struct PendingActivationRecord {
72    format_version: u16,
73    receipt: ActivationReceipt,
74}
75
76impl FileArtifactStore {
77    /// Creates a store rooted at an installation-owned directory.
78    pub fn new(root: impl Into<PathBuf>) -> Self {
79        Self { root: root.into() }
80    }
81
82    /// Returns the artifact file retained for a build.
83    pub fn artifact_path(&self, build_id: &BuildId) -> PathBuf {
84        self.root
85            .join("artifacts")
86            .join(format!("{}.artifact", build_id.as_str()))
87    }
88
89    /// Returns the private path for a staged artifact.
90    pub fn staged_artifact_path(&self, staged: &StagedArtifact) -> PathBuf {
91        self.staged_path(staged.descriptor.build_id())
92    }
93
94    /// Returns a durable activation awaiting supervisor health verification.
95    pub fn pending_activation_receipt(&self) -> UpdateResult<Option<ActivationReceipt>> {
96        self.read_pending_activation()
97    }
98
99    fn staged_path(&self, build_id: &BuildId) -> PathBuf {
100        self.root
101            .join("staged")
102            .join(format!("{}.artifact", build_id.as_str()))
103    }
104
105    fn active_pointer(&self) -> PathBuf {
106        self.root.join("active.json")
107    }
108
109    fn previous_pointer(&self) -> PathBuf {
110        self.root.join("previous.json")
111    }
112
113    fn pending_activation(&self) -> PathBuf {
114        self.root.join("pending-activation.json")
115    }
116
117    fn initialize(&self) -> UpdateResult<()> {
118        fs::create_dir_all(self.root.join("artifacts"))
119            .and_then(|_| fs::create_dir_all(self.root.join("staged")))
120            .map_err(|error| UpdateError::Store(error.to_string()))?;
121        reject_directory(&self.root)?;
122        reject_directory(&self.root.join("artifacts"))?;
123        reject_directory(&self.root.join("staged"))
124    }
125
126    fn read_pointer(&self, path: &Path) -> UpdateResult<Option<ArtifactDescriptor>> {
127        let bytes = match read_regular_file_bounded(path, 1_048_576) {
128            Ok(bytes) => bytes,
129            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
130            Err(error) => return Err(UpdateError::Store(error.to_string())),
131        };
132        let pointer: ArtifactPointer = serde_json::from_slice(&bytes)
133            .map_err(|error| UpdateError::Store(error.to_string()))?;
134        if pointer.format_version != UPDATE_METADATA_FORMAT_VERSION {
135            return Err(UpdateError::Store(
136                "unsupported artifact pointer format".to_string(),
137            ));
138        }
139        pointer.descriptor.validate()?;
140        Ok(Some(pointer.descriptor))
141    }
142
143    fn write_pointer(&self, path: &Path, descriptor: &ArtifactDescriptor) -> UpdateResult<()> {
144        let pointer = ArtifactPointer {
145            format_version: UPDATE_METADATA_FORMAT_VERSION,
146            descriptor: descriptor.clone(),
147        };
148        let bytes = serde_json::to_vec_pretty(&pointer)
149            .map_err(|error| UpdateError::Store(error.to_string()))?;
150        atomic_write(path, &bytes)
151    }
152}
153
154impl ArtifactStore for FileArtifactStore {
155    fn recover(&self) -> UpdateResult<()> {
156        let Some(receipt) = self.read_pending_activation()? else {
157            remove_if_exists(&self.previous_pointer())?;
158            return Ok(());
159        };
160        match self.current()? {
161            Some(current) if current.build_id() == receipt.activated.build_id() => {
162                self.rollback(&receipt)
163            }
164            _ => {
165                remove_if_exists(&self.previous_pointer())?;
166                remove_if_exists(&self.pending_activation())
167            }
168        }
169    }
170
171    fn current(&self) -> UpdateResult<Option<ArtifactDescriptor>> {
172        self.read_pointer(&self.active_pointer())
173    }
174
175    fn stage(&self, descriptor: &ArtifactDescriptor, bytes: &[u8]) -> UpdateResult<StagedArtifact> {
176        self.initialize()?;
177        if bytes.len() as u64 != descriptor.size_bytes() || sha256_hex(bytes) != descriptor.sha256()
178        {
179            return Err(UpdateError::ChecksumMismatch);
180        }
181        let path = self.staged_path(descriptor.build_id());
182        atomic_write(&path, bytes)?;
183        Ok(StagedArtifact {
184            descriptor: descriptor.clone(),
185            staging_reference: path.to_string_lossy().into_owned(),
186        })
187    }
188
189    fn discard_staged(&self, staged: &StagedArtifact) -> UpdateResult<()> {
190        let expected = self.staged_path(staged.descriptor.build_id());
191        if staged.staging_reference != expected.to_string_lossy() {
192            return Err(UpdateError::Store(
193                "staged artifact reference does not belong to this store".to_string(),
194            ));
195        }
196        remove_if_exists(&expected)
197    }
198
199    fn activate(&self, staged: StagedArtifact) -> UpdateResult<ActivationReceipt> {
200        self.activate_inner(staged, None)
201    }
202
203    fn rollback(&self, receipt: &ActivationReceipt) -> UpdateResult<()> {
204        let current = self.current()?.ok_or_else(|| {
205            UpdateError::Store("cannot rollback without an active artifact".to_string())
206        })?;
207        if current.build_id() != receipt.activated.build_id() {
208            return Err(UpdateError::Store(
209                "active artifact changed after activation".to_string(),
210            ));
211        }
212        match &receipt.previous {
213            Some(previous) => self.write_pointer(&self.active_pointer(), previous)?,
214            None => remove_if_exists(&self.active_pointer())?,
215        }
216        remove_if_exists(&self.previous_pointer())?;
217        remove_if_exists(&self.pending_activation())
218    }
219
220    fn commit(&self, receipt: &ActivationReceipt) -> UpdateResult<()> {
221        let current = self.current()?.ok_or_else(|| {
222            UpdateError::Store("cannot commit without an active artifact".to_string())
223        })?;
224        if current.build_id() != receipt.activated.build_id() {
225            return Err(UpdateError::Store(
226                "active artifact changed before commit".to_string(),
227            ));
228        }
229        remove_if_exists(&self.previous_pointer())?;
230        remove_if_exists(&self.pending_activation())
231    }
232}
233
234#[cfg_attr(not(test), allow(dead_code))]
235#[derive(Debug, Clone, Copy, PartialEq, Eq)]
236pub(crate) enum StoreFaultPoint {
237    ArtifactMoved,
238    PreviousPointerWritten,
239    PendingReceiptWritten,
240    ActivePointerWritten,
241}
242
243impl FileArtifactStore {
244    fn activate_inner(
245        &self,
246        staged: StagedArtifact,
247        fault: Option<StoreFaultPoint>,
248    ) -> UpdateResult<ActivationReceipt> {
249        self.initialize()?;
250        let expected = self.staged_path(staged.descriptor.build_id());
251        if Path::new(&staged.staging_reference) != expected {
252            return Err(UpdateError::Store(
253                "staging reference does not belong to this store".to_string(),
254            ));
255        }
256        let previous = self.current()?;
257        let artifact_path = self.artifact_path(staged.descriptor.build_id());
258        verify_artifact(&expected, &staged.descriptor)?;
259        install_artifact(&expected, &artifact_path, &staged.descriptor)?;
260        sync_parent_directory(
261            artifact_path
262                .parent()
263                .ok_or_else(|| UpdateError::Store("artifact path has no parent".to_string()))?,
264        )?;
265        inject_store_fault(fault, StoreFaultPoint::ArtifactMoved)?;
266        if let Some(previous) = &previous {
267            self.write_pointer(&self.previous_pointer(), previous)?;
268        } else {
269            remove_if_exists(&self.previous_pointer())?;
270        }
271        inject_store_fault(fault, StoreFaultPoint::PreviousPointerWritten)?;
272        let receipt = ActivationReceipt {
273            activated: staged.descriptor.clone(),
274            previous,
275        };
276        self.write_pending_activation(&receipt)?;
277        inject_store_fault(fault, StoreFaultPoint::PendingReceiptWritten)?;
278        self.write_pointer(&self.active_pointer(), &staged.descriptor)?;
279        inject_store_fault(fault, StoreFaultPoint::ActivePointerWritten)?;
280        Ok(receipt)
281    }
282
283    #[cfg(test)]
284    pub(crate) fn activate_with_fault(
285        &self,
286        staged: StagedArtifact,
287        fault: StoreFaultPoint,
288    ) -> UpdateResult<ActivationReceipt> {
289        self.activate_inner(staged, Some(fault))
290    }
291
292    fn read_pending_activation(&self) -> UpdateResult<Option<ActivationReceipt>> {
293        let bytes = match read_bounded(&self.pending_activation(), 1_048_576)? {
294            Some(bytes) => bytes,
295            None => return Ok(None),
296        };
297        let record = serde_json::from_slice::<PendingActivationRecord>(&bytes)
298            .map_err(|_| UpdateError::Store("NO MORE SUPPORTED PLEASE UPDATE".to_string()))?;
299        if record.format_version != UPDATE_METADATA_FORMAT_VERSION {
300            return Err(UpdateError::Store(
301                "NO MORE SUPPORTED PLEASE UPDATE".to_string(),
302            ));
303        }
304        let receipt = record.receipt;
305        receipt.activated.validate()?;
306        if let Some(previous) = &receipt.previous {
307            previous.validate()?;
308        }
309        Ok(Some(receipt))
310    }
311
312    fn write_pending_activation(&self, receipt: &ActivationReceipt) -> UpdateResult<()> {
313        let record = PendingActivationRecord {
314            format_version: UPDATE_METADATA_FORMAT_VERSION,
315            receipt: receipt.clone(),
316        };
317        let bytes = serde_json::to_vec_pretty(&record)
318            .map_err(|error| UpdateError::Store(error.to_string()))?;
319        atomic_write(&self.pending_activation(), &bytes)
320    }
321}
322
323fn read_bounded(path: &Path, max_bytes: usize) -> UpdateResult<Option<Vec<u8>>> {
324    let bytes = match read_regular_file_bounded(path, max_bytes) {
325        Ok(bytes) => bytes,
326        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
327        Err(error) => return Err(UpdateError::Store(error.to_string())),
328    };
329    Ok(Some(bytes))
330}
331
332fn verify_artifact(path: &Path, descriptor: &ArtifactDescriptor) -> UpdateResult<()> {
333    let max_bytes = usize::try_from(descriptor.size_bytes())
334        .map_err(|_| UpdateError::Store("artifact size exceeds this platform".to_string()))?;
335    let bytes = read_regular_file_bounded(path, max_bytes)
336        .map_err(|error| UpdateError::Store(error.to_string()))?;
337    if bytes.len() as u64 != descriptor.size_bytes() || sha256_hex(&bytes) != descriptor.sha256() {
338        return Err(UpdateError::ChecksumMismatch);
339    }
340    Ok(())
341}
342
343fn install_artifact(
344    staged_path: &Path,
345    artifact_path: &Path,
346    descriptor: &ArtifactDescriptor,
347) -> UpdateResult<()> {
348    match fs::hard_link(staged_path, artifact_path) {
349        Ok(()) => {}
350        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
351            verify_artifact(artifact_path, descriptor)?;
352        }
353        Err(error) => return Err(UpdateError::Store(error.to_string())),
354    }
355    remove_if_exists(staged_path)
356}
357
358fn atomic_write(path: &Path, bytes: &[u8]) -> UpdateResult<()> {
359    let parent = path
360        .parent()
361        .ok_or_else(|| UpdateError::Store("path has no parent".to_string()))?;
362    fs::create_dir_all(parent).map_err(|error| UpdateError::Store(error.to_string()))?;
363    reject_directory(parent)?;
364    reject_optional_regular_file(path)?;
365    let temporary = path.with_extension(format!(
366        "tmp-{}-{}",
367        std::process::id(),
368        UPDATE_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed)
369    ));
370    let result = (|| {
371        let mut file = OpenOptions::new()
372            .create_new(true)
373            .write(true)
374            .open(&temporary)
375            .map_err(|error| UpdateError::Store(error.to_string()))?;
376        file.write_all(bytes)
377            .and_then(|_| file.sync_all())
378            .map_err(|error| UpdateError::Store(error.to_string()))?;
379        fs::rename(&temporary, path).map_err(|error| UpdateError::Store(error.to_string()))?;
380        sync_parent_directory(parent)
381    })();
382    if result.is_err() {
383        let _ = fs::remove_file(temporary);
384    }
385    result
386}
387
388#[cfg(unix)]
389fn sync_parent_directory(path: &Path) -> UpdateResult<()> {
390    File::open(path)
391        .and_then(|directory| directory.sync_all())
392        .map_err(|error| UpdateError::Store(error.to_string()))
393}
394
395#[cfg(not(unix))]
396fn sync_parent_directory(_path: &Path) -> UpdateResult<()> {
397    Ok(())
398}
399
400fn remove_if_exists(path: &Path) -> UpdateResult<()> {
401    match fs::remove_file(path) {
402        Ok(()) => path
403            .parent()
404            .map(sync_parent_directory)
405            .transpose()
406            .map(|_| ()),
407        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
408        Err(error) => Err(UpdateError::Store(error.to_string())),
409    }
410}
411
412fn inject_store_fault(
413    actual: Option<StoreFaultPoint>,
414    expected: StoreFaultPoint,
415) -> UpdateResult<()> {
416    if actual == Some(expected) {
417        return Err(UpdateError::Store(format!(
418            "injected store fault at {expected:?}"
419        )));
420    }
421    Ok(())
422}
423
424fn reject_optional_regular_file(path: &Path) -> UpdateResult<()> {
425    match fs::symlink_metadata(path) {
426        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => Err(
427            UpdateError::Store("update path is not a regular file".to_string()),
428        ),
429        Ok(_) => Ok(()),
430        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
431        Err(error) => Err(UpdateError::Store(error.to_string())),
432    }
433}
434
435fn reject_directory(path: &Path) -> UpdateResult<()> {
436    let metadata =
437        fs::symlink_metadata(path).map_err(|error| UpdateError::Store(error.to_string()))?;
438    if metadata.file_type().is_symlink() || !metadata.is_dir() {
439        return Err(UpdateError::Store(
440            "update root is not a regular directory".to_string(),
441        ));
442    }
443    Ok(())
444}
appcore_update/store.rs

appcore_update/
store.rs
