hyphae_storage/
data_directory.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 fs4::{FileExt, TryLockError};
10use hyphae_core::DISK_FORMAT_VERSION;
11use thiserror::Error;
12
13use crate::{DurableLog, LogError, ManifestError, OpenedLog, manifest::StorageManifest};
14
15const FORMAT_PREFIX: &str = "hyphae-disk-format=";
16const REQUIRED_DIRECTORIES: [&str; 6] = ["manifest", "log", "snapshots", "indexes", "blobs", "tmp"];
17
18/// Failure while opening or initializing a Hyphae data directory.
19#[derive(Debug, Error)]
20pub enum DataDirectoryError {
21    /// Another writer already owns the directory lock.
22    #[error("data directory is already locked by another writer: {0}")]
23    AlreadyLocked(PathBuf),
24
25    /// The `FORMAT` marker does not match the canonical representation.
26    #[error("malformed data format marker: {0}")]
27    MalformedFormat(PathBuf),
28
29    /// The directory was created by a newer, unsupported disk format.
30    #[error("unsupported disk format {found}; this binary supports {supported}")]
31    UnsupportedFormat {
32        /// Version found in `FORMAT`.
33        found: u16,
34        /// Highest version understood by this binary.
35        supported: u16,
36    },
37
38    /// The immutable storage manifest could not be loaded or initialized.
39    #[error(transparent)]
40    Manifest(#[from] ManifestError),
41
42    /// A filesystem operation failed.
43    #[error("failed to {action} {path}: {source}")]
44    Io {
45        /// Operation that failed.
46        action: &'static str,
47        /// Path involved in the operation.
48        path: PathBuf,
49        /// Underlying operating-system error.
50        #[source]
51        source: io::Error,
52    },
53}
54
55/// An exclusively owned Hyphae data directory.
56///
57/// The operating-system lock is held until this value is dropped. Opening the
58/// same directory for a second writer fails instead of relying on cooperative
59/// process behavior.
60#[derive(Debug)]
61pub struct DataDirectory {
62    root: PathBuf,
63    lock: File,
64    manifest: StorageManifest,
65}
66
67impl DataDirectory {
68    /// Opens an existing data directory or initializes a new one.
69    ///
70    /// # Errors
71    ///
72    /// Returns an error when the directory cannot be initialized, is owned by
73    /// another writer, has a malformed marker, or uses a future format.
74    pub fn open(path: impl AsRef<Path>) -> Result<Self, DataDirectoryError> {
75        let root = path.as_ref().to_path_buf();
76        fs::create_dir_all(&root).map_err(|source| DataDirectoryError::Io {
77            action: "create data directory",
78            path: root.clone(),
79            source,
80        })?;
81
82        let lock_path = root.join("LOCK");
83        let lock = OpenOptions::new()
84            .create(true)
85            .read(true)
86            .write(true)
87            .truncate(false)
88            .open(&lock_path)
89            .map_err(|source| DataDirectoryError::Io {
90                action: "open lock file",
91                path: lock_path.clone(),
92                source,
93            })?;
94
95        match FileExt::try_lock(&lock) {
96            Ok(()) => {}
97            Err(TryLockError::WouldBlock) => {
98                return Err(DataDirectoryError::AlreadyLocked(root));
99            }
100            Err(TryLockError::Error(source)) => {
101                return Err(DataDirectoryError::Io {
102                    action: "lock data directory",
103                    path: lock_path,
104                    source,
105                });
106            }
107        }
108
109        initialize_or_validate_format(&root)?;
110        for name in REQUIRED_DIRECTORIES {
111            let directory = root.join(name);
112            fs::create_dir_all(&directory).map_err(|source| DataDirectoryError::Io {
113                action: "create data subdirectory",
114                path: directory,
115                source,
116            })?;
117        }
118
119        let manifest = StorageManifest::load_or_initialize(&root)?;
120
121        Ok(Self {
122            root,
123            lock,
124            manifest,
125        })
126    }
127
128    /// Returns the canonical root path.
129    pub fn path(&self) -> &Path {
130        &self.root
131    }
132
133    /// Returns the path of the active append-only log segment.
134    pub fn active_log_path(&self) -> PathBuf {
135        self.log_path(self.manifest.active_segment)
136    }
137
138    /// Opens the initial durable log while borrowing this directory lock.
139    ///
140    /// The returned writer cannot outlive the exclusive data-directory owner.
141    ///
142    /// # Errors
143    ///
144    /// Returns an error for I/O failures or any complete invalid frame.
145    pub fn open_log(&self) -> Result<OpenedLog<'_>, LogError> {
146        let (log, recovery) = DurableLog::open_file_at(
147            self.active_log_path(),
148            self.manifest.base_sequence,
149            self.manifest.base_digest,
150        )?;
151        Ok(OpenedLog::new(log, recovery))
152    }
153
154    pub(crate) fn log_anchor(&self) -> (u64, [u8; 32]) {
155        (self.manifest.base_sequence, self.manifest.base_digest)
156    }
157
158    pub(crate) fn manifest(&self) -> StorageManifest {
159        self.manifest
160    }
161
162    pub(crate) fn log_path(&self, segment: u64) -> PathBuf {
163        self.root.join("log").join(format!("{segment:020}.hylog"))
164    }
165
166    pub(crate) fn snapshot_path(&self, sequence: u64) -> PathBuf {
167        self.root
168            .join("snapshots")
169            .join(format!("snapshot-{sequence:020}.hysnap"))
170    }
171
172    pub(crate) fn commit_manifest(
173        &mut self,
174        manifest: StorageManifest,
175    ) -> Result<(), DataDirectoryError> {
176        manifest.write_new(&self.root)?;
177        self.manifest = manifest;
178        Ok(())
179    }
180
181    pub(crate) fn cleanup_retired_logs(&self) -> bool {
182        let log_directory = self.root.join("log");
183        let Ok(entries) = fs::read_dir(&log_directory) else {
184            return false;
185        };
186        #[cfg(unix)]
187        let mut removed = false;
188        for entry in entries.flatten() {
189            let path = entry.path();
190            let Some(segment) = segment_from_path(&path) else {
191                continue;
192            };
193            if segment < self.manifest.active_segment {
194                match fs::remove_file(&path) {
195                    Ok(()) => {
196                        #[cfg(unix)]
197                        {
198                            removed = true;
199                        }
200                    }
201                    Err(source) if source.kind() == io::ErrorKind::NotFound => {}
202                    Err(_) => return false,
203                }
204            }
205        }
206        #[cfg(unix)]
207        if removed && sync_directory(&log_directory).is_err() {
208            return false;
209        }
210        true
211    }
212}
213
214impl Drop for DataDirectory {
215    fn drop(&mut self) {
216        let _ignored = FileExt::unlock(&self.lock);
217    }
218}
219
220fn initialize_or_validate_format(root: &Path) -> Result<(), DataDirectoryError> {
221    let format_path = root.join("FORMAT");
222    if format_path.exists() {
223        return validate_format(&format_path);
224    }
225
226    let temporary_path = root.join("FORMAT.new");
227    let mut file = OpenOptions::new()
228        .create(true)
229        .write(true)
230        .truncate(true)
231        .open(&temporary_path)
232        .map_err(|source| DataDirectoryError::Io {
233            action: "create temporary format marker",
234            path: temporary_path.clone(),
235            source,
236        })?;
237    let marker = format!("{FORMAT_PREFIX}{DISK_FORMAT_VERSION}\n");
238    file.write_all(marker.as_bytes())
239        .and_then(|()| file.sync_all())
240        .map_err(|source| DataDirectoryError::Io {
241            action: "initialize temporary format marker",
242            path: temporary_path.clone(),
243            source,
244        })?;
245    drop(file);
246    fs::rename(&temporary_path, &format_path).map_err(|source| DataDirectoryError::Io {
247        action: "promote format marker",
248        path: format_path,
249        source,
250    })?;
251    #[cfg(unix)]
252    sync_directory(root)?;
253    Ok(())
254}
255
256#[cfg(unix)]
257fn sync_directory(path: &Path) -> Result<(), DataDirectoryError> {
258    File::open(path)
259        .and_then(|directory| directory.sync_all())
260        .map_err(|source| DataDirectoryError::Io {
261            action: "synchronize data directory",
262            path: path.to_path_buf(),
263            source,
264        })
265}
266
267fn validate_format(path: &Path) -> Result<(), DataDirectoryError> {
268    let mut marker = String::new();
269    File::open(path)
270        .and_then(|mut file| file.read_to_string(&mut marker))
271        .map_err(|source| DataDirectoryError::Io {
272            action: "read format marker",
273            path: path.to_path_buf(),
274            source,
275        })?;
276
277    let Some(raw_version) = marker
278        .strip_prefix(FORMAT_PREFIX)
279        .and_then(|value| value.strip_suffix('\n'))
280    else {
281        return Err(DataDirectoryError::MalformedFormat(path.to_path_buf()));
282    };
283    let version = raw_version
284        .parse::<u16>()
285        .map_err(|_| DataDirectoryError::MalformedFormat(path.to_path_buf()))?;
286    if version > DISK_FORMAT_VERSION {
287        return Err(DataDirectoryError::UnsupportedFormat {
288            found: version,
289            supported: DISK_FORMAT_VERSION,
290        });
291    }
292    if version != DISK_FORMAT_VERSION {
293        return Err(DataDirectoryError::MalformedFormat(path.to_path_buf()));
294    }
295    Ok(())
296}
297
298fn segment_from_path(path: &Path) -> Option<u64> {
299    let filename = path.file_name()?.to_str()?;
300    let raw_segment = filename.strip_suffix(".hylog")?;
301    if raw_segment.len() != 20 || !raw_segment.bytes().all(|byte| byte.is_ascii_digit()) {
302        return None;
303    }
304    let segment = raw_segment.parse().ok()?;
305    (format!("{segment:020}.hylog") == filename).then_some(segment)
306}
307
308#[cfg(test)]
309mod tests {
310    use std::{error::Error, fs};
311
312    use uuid::Uuid;
313
314    use super::{DataDirectory, DataDirectoryError, REQUIRED_DIRECTORIES};
315    use crate::{DurableLog, test_support::TestDirectory};
316
317    #[test]
318    fn initializes_canonical_layout() -> Result<(), Box<dyn Error>> {
319        let temporary = TestDirectory::new("data-layout")?;
320        let root = temporary.path().join("data");
321        let directory = DataDirectory::open(&root)?;
322
323        assert_eq!(directory.path(), root);
324        assert_eq!(
325            fs::read_to_string(root.join("FORMAT"))?,
326            "hyphae-disk-format=1\n"
327        );
328        assert!(root.join("LOCK").is_file());
329        for name in REQUIRED_DIRECTORIES {
330            assert!(root.join(name).is_dir());
331        }
332        Ok(())
333    }
334
335    #[test]
336    fn rejects_a_second_writer() -> Result<(), Box<dyn Error>> {
337        let temporary = TestDirectory::new("data-lock")?;
338        let first = DataDirectory::open(temporary.path())?;
339        let second = DataDirectory::open(temporary.path());
340
341        assert!(matches!(second, Err(DataDirectoryError::AlreadyLocked(_))));
342        drop(first);
343        DataDirectory::open(temporary.path())?;
344        Ok(())
345    }
346
347    #[test]
348    fn rejects_future_format() -> Result<(), Box<dyn Error>> {
349        let temporary = TestDirectory::new("future-format")?;
350        fs::write(temporary.path().join("FORMAT"), "hyphae-disk-format=2\n")?;
351
352        let result = DataDirectory::open(temporary.path());
353        assert!(matches!(
354            result,
355            Err(DataDirectoryError::UnsupportedFormat {
356                found: 2,
357                supported: 1
358            })
359        ));
360        Ok(())
361    }
362
363    #[test]
364    fn durable_log_cannot_outlive_directory_lock() -> Result<(), Box<dyn Error>> {
365        let temporary = TestDirectory::new("locked-log")?;
366        let directory = DataDirectory::open(temporary.path())?;
367        let mut opened = directory.open_log()?;
368        opened
369            .log
370            .append_transaction(Uuid::now_v7(), &[b"operation".to_vec()])?;
371
372        assert_eq!(opened.recovery.transactions.len(), 0);
373        assert!(matches!(
374            DataDirectory::open(temporary.path()),
375            Err(DataDirectoryError::AlreadyLocked(_))
376        ));
377        Ok(())
378    }
379
380    #[test]
381    fn migrates_an_existing_format_one_directory_without_a_manifest() -> Result<(), Box<dyn Error>>
382    {
383        let temporary = TestDirectory::new("data-manifest-migration")?;
384        let root = temporary.path().join("data");
385        fs::create_dir_all(root.join("log"))?;
386        fs::write(root.join("FORMAT"), "hyphae-disk-format=1\n")?;
387        let log_path = root.join("log/00000000000000000001.hylog");
388        let (mut legacy_log, _) = DurableLog::open_file(&log_path)?;
389        legacy_log.append_transaction(Uuid::now_v7(), &[b"preserved".to_vec()])?;
390        drop(legacy_log);
391
392        let directory = DataDirectory::open(&root)?;
393        assert!(
394            root.join("manifest/00000000000000000001.hymanifest")
395                .is_file()
396        );
397        let opened = directory.open_log()?;
398        assert_eq!(opened.recovery.transactions.len(), 1);
399        assert_eq!(opened.recovery.transactions[0].operations[0], b"preserved");
400        Ok(())
401    }
402}
hyphae_storage/data_directory.rs

hyphae_storage/
data_directory.rs
