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

appcore_update/
store.rs
