appcore_update/
store.rs

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

appcore_update/
store.rs
