hyphae_storage/
manifest.rs

1// SPDX-License-Identifier: Apache-2.0
2
3use std::{
4    fs::{self, File, OpenOptions},
5    io::{self, Read, Write},
6    path::{Path, PathBuf},
7};
8
9use hyphae_core::{DISK_FORMAT_VERSION, MIN_DISK_FORMAT_VERSION};
10use thiserror::Error;
11
12use crate::{
13    StorageLimitError,
14    limits::{OperationDeadline, limit_io_error},
15};
16
17const MAGIC: [u8; 8] = *b"HYMNFST1";
18const MANIFEST_FORMAT_VERSION: u16 = 1;
19const ENCODED_LENGTH: usize = 140;
20const ENCODED_LENGTH_U64: u64 = 140;
21const CHECKSUM_PREFIX_LENGTH: usize = 104;
22const DIGEST_PREFIX_LENGTH: usize = 108;
23const MANIFEST_EXTENSION: &str = "hymanifest";
24
25/// Failure while loading or atomically creating a storage manifest.
26#[derive(Debug, Error)]
27pub enum ManifestError {
28    /// A filesystem operation failed.
29    #[error(transparent)]
30    Io(#[from] io::Error),
31
32    /// A committed manifest violates the canonical representation.
33    #[error("invalid storage manifest {path}: {reason}")]
34    Invalid {
35        /// Invalid manifest path.
36        path: PathBuf,
37        /// Stable diagnostic reason.
38        reason: &'static str,
39    },
40
41    /// The manifest uses a future format.
42    #[error(
43        "unsupported storage manifest version {found}; supported manifest version is {supported}"
44    )]
45    UnsupportedVersion {
46        /// Version found on disk.
47        found: u16,
48        /// Highest version understood by this binary.
49        supported: u16,
50    },
51
52    /// An immutable generation already exists with different content.
53    #[error("storage manifest generation {generation} already exists with different content")]
54    GenerationConflict {
55        /// Conflicting manifest generation.
56        generation: u64,
57    },
58}
59
60impl From<StorageLimitError> for ManifestError {
61    fn from(source: StorageLimitError) -> Self {
62        Self::Io(limit_io_error(source))
63    }
64}
65
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67pub(crate) struct StorageManifest {
68    pub(crate) generation: u64,
69    pub(crate) active_segment: u64,
70    pub(crate) base_sequence: u64,
71    pub(crate) base_digest: [u8; 32],
72    pub(crate) snapshot_digest: [u8; 32],
73}
74
75impl StorageManifest {
76    fn initial() -> Self {
77        Self {
78            generation: 1,
79            active_segment: 1,
80            base_sequence: 0,
81            base_digest: [0; 32],
82            snapshot_digest: [0; 32],
83        }
84    }
85
86    #[cfg(test)]
87    pub(crate) fn load_or_initialize(root: &Path) -> Result<Self, ManifestError> {
88        let deadline = OperationDeadline::new(std::time::Duration::from_secs(60));
89        Self::load_or_initialize_with_limits(root, 1_000_000, &deadline)
90    }
91
92    pub(crate) fn load_or_initialize_with_limits(
93        root: &Path,
94        max_directory_entries: u64,
95        deadline: &OperationDeadline,
96    ) -> Result<Self, ManifestError> {
97        deadline.check()?;
98        let directory = root.join("manifest");
99        let mut generations = Vec::new();
100        let mut entry_count = 0_u64;
101        for entry in fs::read_dir(&directory)? {
102            deadline.check()?;
103            entry_count =
104                entry_count
105                    .checked_add(1)
106                    .ok_or(StorageLimitError::DirectoryEntriesExceeded {
107                        maximum: max_directory_entries,
108                    })?;
109            if entry_count > max_directory_entries {
110                return Err(StorageLimitError::DirectoryEntriesExceeded {
111                    maximum: max_directory_entries,
112                }
113                .into());
114            }
115            let path = entry?.path();
116            if let Some(generation) = generation_from_path(&path)? {
117                generations.push((generation, path));
118            }
119        }
120        generations.sort_unstable_by_key(|(generation, _)| *generation);
121        if let Some((generation, path)) = generations.last() {
122            return decode_manifest(path, *generation);
123        }
124
125        let required_entries =
126            entry_count
127                .checked_add(1)
128                .ok_or(StorageLimitError::DirectoryEntriesExceeded {
129                    maximum: max_directory_entries,
130                })?;
131        if required_entries > max_directory_entries {
132            return Err(StorageLimitError::DirectoryEntriesExceeded {
133                maximum: max_directory_entries,
134            }
135            .into());
136        }
137        let initial = Self::initial();
138        initial.write_new(root)?;
139        Ok(initial)
140    }
141
142    pub(crate) fn write_new(&self, root: &Path) -> Result<(), ManifestError> {
143        self.write_new_inner(root, false)
144    }
145
146    #[cfg(test)]
147    fn write_new_with_injected_temporary_failure(&self, root: &Path) -> Result<(), ManifestError> {
148        self.write_new_inner(root, true)
149    }
150
151    fn write_new_inner(
152        &self,
153        root: &Path,
154        inject_temporary_failure: bool,
155    ) -> Result<(), ManifestError> {
156        validate_semantics(self, &root.join(manifest_filename(self.generation)))?;
157        let manifest_directory = root.join("manifest");
158        let final_path = manifest_directory.join(manifest_filename(self.generation));
159        if final_path.exists() {
160            let existing = decode_manifest(&final_path, self.generation)?;
161            return if existing == *self {
162                Ok(())
163            } else {
164                Err(ManifestError::GenerationConflict {
165                    generation: self.generation,
166                })
167            };
168        }
169
170        let temporary_path = root.join("tmp").join(format!(
171            "manifest-{:020}-{}.tmp",
172            self.generation,
173            uuid::Uuid::now_v7()
174        ));
175        let mut temporary_guard = TemporaryManifestGuard::new(temporary_path.clone());
176        let encoded = encode_manifest(self);
177        let mut file = OpenOptions::new()
178            .create_new(true)
179            .write(true)
180            .open(&temporary_path)?;
181        file.write_all(&encoded)?;
182        file.sync_all()?;
183        drop(file);
184        if inject_temporary_failure {
185            return Err(io::Error::other("injected temporary manifest failure").into());
186        }
187        fs::rename(&temporary_path, &final_path)?;
188        temporary_guard.disarm();
189        #[cfg(unix)]
190        sync_directory(&manifest_directory)?;
191        Ok(())
192    }
193
194    pub(crate) fn path(&self, root: &Path) -> PathBuf {
195        root.join("manifest")
196            .join(manifest_filename(self.generation))
197    }
198}
199
200struct TemporaryManifestGuard {
201    path: PathBuf,
202    armed: bool,
203}
204
205impl TemporaryManifestGuard {
206    fn new(path: PathBuf) -> Self {
207        Self { path, armed: true }
208    }
209
210    fn disarm(&mut self) {
211        self.armed = false;
212    }
213}
214
215impl Drop for TemporaryManifestGuard {
216    fn drop(&mut self) {
217        if self.armed {
218            let _ignored = fs::remove_file(&self.path);
219        }
220    }
221}
222
223fn manifest_filename(generation: u64) -> String {
224    format!("{generation:020}.{MANIFEST_EXTENSION}")
225}
226
227fn generation_from_path(path: &Path) -> Result<Option<u64>, ManifestError> {
228    if path.extension().and_then(|extension| extension.to_str()) != Some(MANIFEST_EXTENSION) {
229        return Ok(None);
230    }
231    let Some(filename) = path.file_name().and_then(|name| name.to_str()) else {
232        return Err(invalid(path, "manifest filename is not UTF-8"));
233    };
234    let Some(raw_generation) = filename.strip_suffix(&format!(".{MANIFEST_EXTENSION}")) else {
235        return Err(invalid(path, "malformed manifest filename"));
236    };
237    let generation = raw_generation
238        .parse::<u64>()
239        .map_err(|_| invalid(path, "malformed manifest generation"))?;
240    if manifest_filename(generation) != filename {
241        return Err(invalid(path, "noncanonical manifest filename"));
242    }
243    Ok(Some(generation))
244}
245
246fn encode_manifest(manifest: &StorageManifest) -> [u8; ENCODED_LENGTH] {
247    let mut encoded = [0_u8; ENCODED_LENGTH];
248    encoded[..8].copy_from_slice(&MAGIC);
249    encoded[8..10].copy_from_slice(&MANIFEST_FORMAT_VERSION.to_le_bytes());
250    encoded[10..12].copy_from_slice(&DISK_FORMAT_VERSION.to_le_bytes());
251    encoded[12..16].copy_from_slice(&0_u32.to_le_bytes());
252    encoded[16..24].copy_from_slice(&manifest.generation.to_le_bytes());
253    encoded[24..32].copy_from_slice(&manifest.active_segment.to_le_bytes());
254    encoded[32..40].copy_from_slice(&manifest.base_sequence.to_le_bytes());
255    encoded[40..72].copy_from_slice(&manifest.base_digest);
256    encoded[72..104].copy_from_slice(&manifest.snapshot_digest);
257    let checksum = crc32c::crc32c(&encoded[..CHECKSUM_PREFIX_LENGTH]);
258    encoded[104..108].copy_from_slice(&checksum.to_le_bytes());
259    let digest = blake3::hash(&encoded[..DIGEST_PREFIX_LENGTH]);
260    encoded[108..140].copy_from_slice(digest.as_bytes());
261    encoded
262}
263
264fn decode_manifest(
265    path: &Path,
266    filename_generation: u64,
267) -> Result<StorageManifest, ManifestError> {
268    let mut file = File::open(path)?;
269    if file.metadata()?.len() != ENCODED_LENGTH_U64 {
270        return Err(invalid(path, "file length mismatch"));
271    }
272    let mut encoded = [0_u8; ENCODED_LENGTH];
273    file.read_exact(&mut encoded)?;
274    if encoded[..8] != MAGIC {
275        return Err(invalid(path, "bad magic"));
276    }
277    let manifest_version = u16::from_le_bytes(copy_array(&encoded[8..10]));
278    if manifest_version != MANIFEST_FORMAT_VERSION {
279        return Err(ManifestError::UnsupportedVersion {
280            found: manifest_version,
281            supported: MANIFEST_FORMAT_VERSION,
282        });
283    }
284    let disk_format = u16::from_le_bytes(copy_array(&encoded[10..12]));
285    if !(MIN_DISK_FORMAT_VERSION..=DISK_FORMAT_VERSION).contains(&disk_format) {
286        return Err(invalid(path, "disk format mismatch"));
287    }
288    if u32::from_le_bytes(copy_array(&encoded[12..16])) != 0 {
289        return Err(invalid(path, "unsupported flags"));
290    }
291    let expected_checksum = u32::from_le_bytes(copy_array(&encoded[104..108]));
292    if crc32c::crc32c(&encoded[..CHECKSUM_PREFIX_LENGTH]) != expected_checksum {
293        return Err(invalid(path, "CRC32C mismatch"));
294    }
295    let expected_digest: [u8; 32] = copy_array(&encoded[108..140]);
296    if *blake3::hash(&encoded[..DIGEST_PREFIX_LENGTH]).as_bytes() != expected_digest {
297        return Err(invalid(path, "BLAKE3 digest mismatch"));
298    }
299
300    let manifest = StorageManifest {
301        generation: u64::from_le_bytes(copy_array(&encoded[16..24])),
302        active_segment: u64::from_le_bytes(copy_array(&encoded[24..32])),
303        base_sequence: u64::from_le_bytes(copy_array(&encoded[32..40])),
304        base_digest: copy_array(&encoded[40..72]),
305        snapshot_digest: copy_array(&encoded[72..104]),
306    };
307    if manifest.generation != filename_generation {
308        return Err(invalid(path, "filename generation mismatch"));
309    }
310    validate_semantics(&manifest, path)?;
311    Ok(manifest)
312}
313
314fn validate_semantics(manifest: &StorageManifest, path: &Path) -> Result<(), ManifestError> {
315    if manifest.generation == 0 || manifest.active_segment != manifest.generation {
316        return Err(invalid(path, "invalid generation or active segment"));
317    }
318    let empty_anchor = manifest.base_sequence == 0;
319    if empty_anchor != (manifest.base_digest == [0; 32])
320        || empty_anchor != (manifest.snapshot_digest == [0; 32])
321    {
322        return Err(invalid(path, "inconsistent snapshot anchor"));
323    }
324    if manifest.generation == 1 && !empty_anchor {
325        return Err(invalid(path, "initial generation has a snapshot anchor"));
326    }
327    if manifest.generation > 1 && empty_anchor {
328        return Err(invalid(
329            path,
330            "compacted generation lacks a snapshot anchor",
331        ));
332    }
333    Ok(())
334}
335
336fn invalid(path: &Path, reason: &'static str) -> ManifestError {
337    ManifestError::Invalid {
338        path: path.to_path_buf(),
339        reason,
340    }
341}
342
343#[cfg(unix)]
344fn sync_directory(path: &Path) -> Result<(), ManifestError> {
345    File::open(path)?.sync_all()?;
346    Ok(())
347}
348
349fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
350    let mut output = [0_u8; N];
351    output.copy_from_slice(source);
352    output
353}
354
355#[cfg(test)]
356mod tests {
357    use std::{error::Error, fs, io::Write};
358
359    use super::{ManifestError, StorageManifest};
360    use crate::{StorageLimitError, storage_limit_from_io, test_support::TestDirectory};
361
362    fn initialize_layout(root: &std::path::Path) -> Result<(), Box<dyn Error>> {
363        fs::create_dir_all(root.join("manifest"))?;
364        fs::create_dir_all(root.join("tmp"))?;
365        Ok(())
366    }
367
368    #[test]
369    fn initializes_and_reloads_an_immutable_manifest() -> Result<(), Box<dyn Error>> {
370        let temporary = TestDirectory::new("manifest-initial")?;
371        initialize_layout(temporary.path())?;
372
373        let created = StorageManifest::load_or_initialize(temporary.path())?;
374        let reopened = StorageManifest::load_or_initialize(temporary.path())?;
375        assert_eq!(created, reopened);
376        assert_eq!(created.generation, 1);
377        assert_eq!(created.base_sequence, 0);
378        Ok(())
379    }
380
381    #[test]
382    fn ignores_an_interrupted_temporary_manifest() -> Result<(), Box<dyn Error>> {
383        let temporary = TestDirectory::new("manifest-interrupted")?;
384        initialize_layout(temporary.path())?;
385        let mut partial = fs::File::create(temporary.path().join("tmp/manifest-partial.tmp"))?;
386        partial.write_all(b"partial")?;
387        partial.sync_all()?;
388
389        let manifest = StorageManifest::load_or_initialize(temporary.path())?;
390        assert_eq!(manifest.generation, 1);
391        Ok(())
392    }
393
394    #[test]
395    fn rejects_corruption_in_the_latest_generation() -> Result<(), Box<dyn Error>> {
396        let temporary = TestDirectory::new("manifest-corrupt")?;
397        initialize_layout(temporary.path())?;
398        StorageManifest::load_or_initialize(temporary.path())?;
399        let path = temporary
400            .path()
401            .join("manifest/00000000000000000001.hymanifest");
402        let mut bytes = fs::read(&path)?;
403        bytes[40] ^= 1;
404        fs::write(&path, bytes)?;
405
406        assert!(matches!(
407            StorageManifest::load_or_initialize(temporary.path()),
408            Err(ManifestError::Invalid {
409                reason: "CRC32C mismatch",
410                ..
411            })
412        ));
413        Ok(())
414    }
415
416    #[test]
417    fn removes_a_temporary_manifest_after_a_pre_rename_failure() -> Result<(), Box<dyn Error>> {
418        let temporary = TestDirectory::new("manifest-temporary-cleanup")?;
419        initialize_layout(temporary.path())?;
420        let manifest = StorageManifest::initial();
421
422        assert!(matches!(
423            manifest.write_new_with_injected_temporary_failure(temporary.path()),
424            Err(ManifestError::Io(_))
425        ));
426        assert_eq!(fs::read_dir(temporary.path().join("tmp"))?.count(), 0);
427        assert_eq!(fs::read_dir(temporary.path().join("manifest"))?.count(), 0);
428        Ok(())
429    }
430
431    #[test]
432    fn reserves_capacity_before_initializing_a_manifest() -> Result<(), Box<dyn Error>> {
433        let temporary = TestDirectory::new("manifest-initial-capacity")?;
434        initialize_layout(temporary.path())?;
435        fs::write(temporary.path().join("manifest/occupied"), b"occupied")?;
436        let deadline = crate::limits::OperationDeadline::new(std::time::Duration::from_secs(1));
437
438        assert!(matches!(
439            StorageManifest::load_or_initialize_with_limits(temporary.path(), 1, &deadline),
440            Err(ManifestError::Io(source))
441                if matches!(
442                    storage_limit_from_io(&source),
443                    Some(StorageLimitError::DirectoryEntriesExceeded { maximum: 1 })
444                )
445        ));
446        assert!(
447            !temporary
448                .path()
449                .join("manifest/00000000000000000001.hymanifest")
450                .exists()
451        );
452        Ok(())
453    }
454}
hyphae_storage/manifest.rs

hyphae_storage/
manifest.rs
