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

hyphae_storage/
data_directory.rs
