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
12const MAGIC: [u8; 8] = *b"HYMNFST1";
13const MANIFEST_FORMAT_VERSION: u16 = 1;
14const ENCODED_LENGTH: usize = 140;
15const ENCODED_LENGTH_U64: u64 = 140;
16const CHECKSUM_PREFIX_LENGTH: usize = 104;
17const DIGEST_PREFIX_LENGTH: usize = 108;
18const MANIFEST_EXTENSION: &str = "hymanifest";
19
20/// Failure while loading or atomically creating a storage manifest.
21#[derive(Debug, Error)]
22pub enum ManifestError {
23    /// A filesystem operation failed.
24    #[error(transparent)]
25    Io(#[from] io::Error),
26
27    /// A committed manifest violates the canonical representation.
28    #[error("invalid storage manifest {path}: {reason}")]
29    Invalid {
30        /// Invalid manifest path.
31        path: PathBuf,
32        /// Stable diagnostic reason.
33        reason: &'static str,
34    },
35
36    /// The manifest uses a future format.
37    #[error(
38        "unsupported storage manifest version {found}; supported manifest version is {supported}"
39    )]
40    UnsupportedVersion {
41        /// Version found on disk.
42        found: u16,
43        /// Highest version understood by this binary.
44        supported: u16,
45    },
46
47    /// An immutable generation already exists with different content.
48    #[error("storage manifest generation {generation} already exists with different content")]
49    GenerationConflict {
50        /// Conflicting manifest generation.
51        generation: u64,
52    },
53}
54
55#[derive(Clone, Copy, Debug, Eq, PartialEq)]
56pub(crate) struct StorageManifest {
57    pub(crate) generation: u64,
58    pub(crate) active_segment: u64,
59    pub(crate) base_sequence: u64,
60    pub(crate) base_digest: [u8; 32],
61    pub(crate) snapshot_digest: [u8; 32],
62}
63
64impl StorageManifest {
65    fn initial() -> Self {
66        Self {
67            generation: 1,
68            active_segment: 1,
69            base_sequence: 0,
70            base_digest: [0; 32],
71            snapshot_digest: [0; 32],
72        }
73    }
74
75    pub(crate) fn load_or_initialize(root: &Path) -> Result<Self, ManifestError> {
76        let directory = root.join("manifest");
77        let mut generations = Vec::new();
78        for entry in fs::read_dir(&directory)? {
79            let path = entry?.path();
80            if let Some(generation) = generation_from_path(&path)? {
81                generations.push((generation, path));
82            }
83        }
84        generations.sort_unstable_by_key(|(generation, _)| *generation);
85        if let Some((generation, path)) = generations.last() {
86            return decode_manifest(path, *generation);
87        }
88
89        let initial = Self::initial();
90        initial.write_new(root)?;
91        Ok(initial)
92    }
93
94    pub(crate) fn write_new(&self, root: &Path) -> Result<(), ManifestError> {
95        validate_semantics(self, &root.join(manifest_filename(self.generation)))?;
96        let manifest_directory = root.join("manifest");
97        let final_path = manifest_directory.join(manifest_filename(self.generation));
98        if final_path.exists() {
99            let existing = decode_manifest(&final_path, self.generation)?;
100            return if existing == *self {
101                Ok(())
102            } else {
103                Err(ManifestError::GenerationConflict {
104                    generation: self.generation,
105                })
106            };
107        }
108
109        let temporary_path = root.join("tmp").join(format!(
110            "manifest-{:020}-{}.tmp",
111            self.generation,
112            uuid::Uuid::now_v7()
113        ));
114        let encoded = encode_manifest(self);
115        let mut file = OpenOptions::new()
116            .create_new(true)
117            .write(true)
118            .open(&temporary_path)?;
119        file.write_all(&encoded)?;
120        file.sync_all()?;
121        drop(file);
122        fs::rename(&temporary_path, &final_path)?;
123        #[cfg(unix)]
124        sync_directory(&manifest_directory)?;
125        Ok(())
126    }
127}
128
129fn manifest_filename(generation: u64) -> String {
130    format!("{generation:020}.{MANIFEST_EXTENSION}")
131}
132
133fn generation_from_path(path: &Path) -> Result<Option<u64>, ManifestError> {
134    if path.extension().and_then(|extension| extension.to_str()) != Some(MANIFEST_EXTENSION) {
135        return Ok(None);
136    }
137    let Some(filename) = path.file_name().and_then(|name| name.to_str()) else {
138        return Err(invalid(path, "manifest filename is not UTF-8"));
139    };
140    let Some(raw_generation) = filename.strip_suffix(&format!(".{MANIFEST_EXTENSION}")) else {
141        return Err(invalid(path, "malformed manifest filename"));
142    };
143    let generation = raw_generation
144        .parse::<u64>()
145        .map_err(|_| invalid(path, "malformed manifest generation"))?;
146    if manifest_filename(generation) != filename {
147        return Err(invalid(path, "noncanonical manifest filename"));
148    }
149    Ok(Some(generation))
150}
151
152fn encode_manifest(manifest: &StorageManifest) -> [u8; ENCODED_LENGTH] {
153    let mut encoded = [0_u8; ENCODED_LENGTH];
154    encoded[..8].copy_from_slice(&MAGIC);
155    encoded[8..10].copy_from_slice(&MANIFEST_FORMAT_VERSION.to_le_bytes());
156    encoded[10..12].copy_from_slice(&DISK_FORMAT_VERSION.to_le_bytes());
157    encoded[12..16].copy_from_slice(&0_u32.to_le_bytes());
158    encoded[16..24].copy_from_slice(&manifest.generation.to_le_bytes());
159    encoded[24..32].copy_from_slice(&manifest.active_segment.to_le_bytes());
160    encoded[32..40].copy_from_slice(&manifest.base_sequence.to_le_bytes());
161    encoded[40..72].copy_from_slice(&manifest.base_digest);
162    encoded[72..104].copy_from_slice(&manifest.snapshot_digest);
163    let checksum = crc32c::crc32c(&encoded[..CHECKSUM_PREFIX_LENGTH]);
164    encoded[104..108].copy_from_slice(&checksum.to_le_bytes());
165    let digest = blake3::hash(&encoded[..DIGEST_PREFIX_LENGTH]);
166    encoded[108..140].copy_from_slice(digest.as_bytes());
167    encoded
168}
169
170fn decode_manifest(
171    path: &Path,
172    filename_generation: u64,
173) -> Result<StorageManifest, ManifestError> {
174    let mut file = File::open(path)?;
175    if file.metadata()?.len() != ENCODED_LENGTH_U64 {
176        return Err(invalid(path, "file length mismatch"));
177    }
178    let mut encoded = [0_u8; ENCODED_LENGTH];
179    file.read_exact(&mut encoded)?;
180    if encoded[..8] != MAGIC {
181        return Err(invalid(path, "bad magic"));
182    }
183    let manifest_version = u16::from_le_bytes(copy_array(&encoded[8..10]));
184    if manifest_version != MANIFEST_FORMAT_VERSION {
185        return Err(ManifestError::UnsupportedVersion {
186            found: manifest_version,
187            supported: MANIFEST_FORMAT_VERSION,
188        });
189    }
190    let disk_format = u16::from_le_bytes(copy_array(&encoded[10..12]));
191    if !(MIN_DISK_FORMAT_VERSION..=DISK_FORMAT_VERSION).contains(&disk_format) {
192        return Err(invalid(path, "disk format mismatch"));
193    }
194    if u32::from_le_bytes(copy_array(&encoded[12..16])) != 0 {
195        return Err(invalid(path, "unsupported flags"));
196    }
197    let expected_checksum = u32::from_le_bytes(copy_array(&encoded[104..108]));
198    if crc32c::crc32c(&encoded[..CHECKSUM_PREFIX_LENGTH]) != expected_checksum {
199        return Err(invalid(path, "CRC32C mismatch"));
200    }
201    let expected_digest: [u8; 32] = copy_array(&encoded[108..140]);
202    if *blake3::hash(&encoded[..DIGEST_PREFIX_LENGTH]).as_bytes() != expected_digest {
203        return Err(invalid(path, "BLAKE3 digest mismatch"));
204    }
205
206    let manifest = StorageManifest {
207        generation: u64::from_le_bytes(copy_array(&encoded[16..24])),
208        active_segment: u64::from_le_bytes(copy_array(&encoded[24..32])),
209        base_sequence: u64::from_le_bytes(copy_array(&encoded[32..40])),
210        base_digest: copy_array(&encoded[40..72]),
211        snapshot_digest: copy_array(&encoded[72..104]),
212    };
213    if manifest.generation != filename_generation {
214        return Err(invalid(path, "filename generation mismatch"));
215    }
216    validate_semantics(&manifest, path)?;
217    Ok(manifest)
218}
219
220fn validate_semantics(manifest: &StorageManifest, path: &Path) -> Result<(), ManifestError> {
221    if manifest.generation == 0 || manifest.active_segment != manifest.generation {
222        return Err(invalid(path, "invalid generation or active segment"));
223    }
224    let empty_anchor = manifest.base_sequence == 0;
225    if empty_anchor != (manifest.base_digest == [0; 32])
226        || empty_anchor != (manifest.snapshot_digest == [0; 32])
227    {
228        return Err(invalid(path, "inconsistent snapshot anchor"));
229    }
230    if manifest.generation == 1 && !empty_anchor {
231        return Err(invalid(path, "initial generation has a snapshot anchor"));
232    }
233    if manifest.generation > 1 && empty_anchor {
234        return Err(invalid(
235            path,
236            "compacted generation lacks a snapshot anchor",
237        ));
238    }
239    Ok(())
240}
241
242fn invalid(path: &Path, reason: &'static str) -> ManifestError {
243    ManifestError::Invalid {
244        path: path.to_path_buf(),
245        reason,
246    }
247}
248
249#[cfg(unix)]
250fn sync_directory(path: &Path) -> Result<(), ManifestError> {
251    File::open(path)?.sync_all()?;
252    Ok(())
253}
254
255fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
256    let mut output = [0_u8; N];
257    output.copy_from_slice(source);
258    output
259}
260
261#[cfg(test)]
262mod tests {
263    use std::{error::Error, fs, io::Write};
264
265    use super::{ManifestError, StorageManifest};
266    use crate::test_support::TestDirectory;
267
268    fn initialize_layout(root: &std::path::Path) -> Result<(), Box<dyn Error>> {
269        fs::create_dir_all(root.join("manifest"))?;
270        fs::create_dir_all(root.join("tmp"))?;
271        Ok(())
272    }
273
274    #[test]
275    fn initializes_and_reloads_an_immutable_manifest() -> Result<(), Box<dyn Error>> {
276        let temporary = TestDirectory::new("manifest-initial")?;
277        initialize_layout(temporary.path())?;
278
279        let created = StorageManifest::load_or_initialize(temporary.path())?;
280        let reopened = StorageManifest::load_or_initialize(temporary.path())?;
281        assert_eq!(created, reopened);
282        assert_eq!(created.generation, 1);
283        assert_eq!(created.base_sequence, 0);
284        Ok(())
285    }
286
287    #[test]
288    fn ignores_an_interrupted_temporary_manifest() -> Result<(), Box<dyn Error>> {
289        let temporary = TestDirectory::new("manifest-interrupted")?;
290        initialize_layout(temporary.path())?;
291        let mut partial = fs::File::create(temporary.path().join("tmp/manifest-partial.tmp"))?;
292        partial.write_all(b"partial")?;
293        partial.sync_all()?;
294
295        let manifest = StorageManifest::load_or_initialize(temporary.path())?;
296        assert_eq!(manifest.generation, 1);
297        Ok(())
298    }
299
300    #[test]
301    fn rejects_corruption_in_the_latest_generation() -> Result<(), Box<dyn Error>> {
302        let temporary = TestDirectory::new("manifest-corrupt")?;
303        initialize_layout(temporary.path())?;
304        StorageManifest::load_or_initialize(temporary.path())?;
305        let path = temporary
306            .path()
307            .join("manifest/00000000000000000001.hymanifest");
308        let mut bytes = fs::read(&path)?;
309        bytes[40] ^= 1;
310        fs::write(&path, bytes)?;
311
312        assert!(matches!(
313            StorageManifest::load_or_initialize(temporary.path()),
314            Err(ManifestError::Invalid {
315                reason: "CRC32C mismatch",
316                ..
317            })
318        ));
319        Ok(())
320    }
321}
hyphae_storage/manifest.rs

hyphae_storage/
manifest.rs
