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;
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    if u16::from_le_bytes(copy_array(&encoded[10..12])) != DISK_FORMAT_VERSION {
191        return Err(invalid(path, "disk format mismatch"));
192    }
193    if u32::from_le_bytes(copy_array(&encoded[12..16])) != 0 {
194        return Err(invalid(path, "unsupported flags"));
195    }
196    let expected_checksum = u32::from_le_bytes(copy_array(&encoded[104..108]));
197    if crc32c::crc32c(&encoded[..CHECKSUM_PREFIX_LENGTH]) != expected_checksum {
198        return Err(invalid(path, "CRC32C mismatch"));
199    }
200    let expected_digest: [u8; 32] = copy_array(&encoded[108..140]);
201    if *blake3::hash(&encoded[..DIGEST_PREFIX_LENGTH]).as_bytes() != expected_digest {
202        return Err(invalid(path, "BLAKE3 digest mismatch"));
203    }
204
205    let manifest = StorageManifest {
206        generation: u64::from_le_bytes(copy_array(&encoded[16..24])),
207        active_segment: u64::from_le_bytes(copy_array(&encoded[24..32])),
208        base_sequence: u64::from_le_bytes(copy_array(&encoded[32..40])),
209        base_digest: copy_array(&encoded[40..72]),
210        snapshot_digest: copy_array(&encoded[72..104]),
211    };
212    if manifest.generation != filename_generation {
213        return Err(invalid(path, "filename generation mismatch"));
214    }
215    validate_semantics(&manifest, path)?;
216    Ok(manifest)
217}
218
219fn validate_semantics(manifest: &StorageManifest, path: &Path) -> Result<(), ManifestError> {
220    if manifest.generation == 0 || manifest.active_segment != manifest.generation {
221        return Err(invalid(path, "invalid generation or active segment"));
222    }
223    let empty_anchor = manifest.base_sequence == 0;
224    if empty_anchor != (manifest.base_digest == [0; 32])
225        || empty_anchor != (manifest.snapshot_digest == [0; 32])
226    {
227        return Err(invalid(path, "inconsistent snapshot anchor"));
228    }
229    if manifest.generation == 1 && !empty_anchor {
230        return Err(invalid(path, "initial generation has a snapshot anchor"));
231    }
232    if manifest.generation > 1 && empty_anchor {
233        return Err(invalid(
234            path,
235            "compacted generation lacks a snapshot anchor",
236        ));
237    }
238    Ok(())
239}
240
241fn invalid(path: &Path, reason: &'static str) -> ManifestError {
242    ManifestError::Invalid {
243        path: path.to_path_buf(),
244        reason,
245    }
246}
247
248#[cfg(unix)]
249fn sync_directory(path: &Path) -> Result<(), ManifestError> {
250    File::open(path)?.sync_all()?;
251    Ok(())
252}
253
254fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
255    let mut output = [0_u8; N];
256    output.copy_from_slice(source);
257    output
258}
259
260#[cfg(test)]
261mod tests {
262    use std::{error::Error, fs, io::Write};
263
264    use super::{ManifestError, StorageManifest};
265    use crate::test_support::TestDirectory;
266
267    fn initialize_layout(root: &std::path::Path) -> Result<(), Box<dyn Error>> {
268        fs::create_dir_all(root.join("manifest"))?;
269        fs::create_dir_all(root.join("tmp"))?;
270        Ok(())
271    }
272
273    #[test]
274    fn initializes_and_reloads_an_immutable_manifest() -> Result<(), Box<dyn Error>> {
275        let temporary = TestDirectory::new("manifest-initial")?;
276        initialize_layout(temporary.path())?;
277
278        let created = StorageManifest::load_or_initialize(temporary.path())?;
279        let reopened = StorageManifest::load_or_initialize(temporary.path())?;
280        assert_eq!(created, reopened);
281        assert_eq!(created.generation, 1);
282        assert_eq!(created.base_sequence, 0);
283        Ok(())
284    }
285
286    #[test]
287    fn ignores_an_interrupted_temporary_manifest() -> Result<(), Box<dyn Error>> {
288        let temporary = TestDirectory::new("manifest-interrupted")?;
289        initialize_layout(temporary.path())?;
290        let mut partial = fs::File::create(temporary.path().join("tmp/manifest-partial.tmp"))?;
291        partial.write_all(b"partial")?;
292        partial.sync_all()?;
293
294        let manifest = StorageManifest::load_or_initialize(temporary.path())?;
295        assert_eq!(manifest.generation, 1);
296        Ok(())
297    }
298
299    #[test]
300    fn rejects_corruption_in_the_latest_generation() -> Result<(), Box<dyn Error>> {
301        let temporary = TestDirectory::new("manifest-corrupt")?;
302        initialize_layout(temporary.path())?;
303        StorageManifest::load_or_initialize(temporary.path())?;
304        let path = temporary
305            .path()
306            .join("manifest/00000000000000000001.hymanifest");
307        let mut bytes = fs::read(&path)?;
308        bytes[40] ^= 1;
309        fs::write(&path, bytes)?;
310
311        assert!(matches!(
312            StorageManifest::load_or_initialize(temporary.path()),
313            Err(ManifestError::Invalid {
314                reason: "CRC32C mismatch",
315                ..
316            })
317        ));
318        Ok(())
319    }
320}
hyphae_storage/manifest.rs

hyphae_storage/
manifest.rs
