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::{
14    DurableLog, LogError, ManifestError, OpenedLog, RecoveryLimits, StorageLimitError,
15    limits::{OperationDeadline, limit_io_error},
16    manifest::StorageManifest,
17};
18
19const FORMAT_PREFIX: &str = "hyphae-disk-format=";
20const MAX_FORMAT_MARKER_BYTES: usize = 64;
21const FORMAT_MARKER_READ_LIMIT: u64 = 65;
22const REQUIRED_DIRECTORIES: [&str; 6] = ["manifest", "log", "snapshots", "indexes", "blobs", "tmp"];
23
24/// Failure while opening or initializing a Hyphae data directory.
25#[derive(Debug, Error)]
26pub enum DataDirectoryError {
27    /// Another writer already owns the directory lock.
28    #[error("data directory is already locked by another writer: {0}")]
29    AlreadyLocked(PathBuf),
30
31    /// The `FORMAT` marker does not match the canonical representation.
32    #[error("malformed data format marker: {0}")]
33    MalformedFormat(PathBuf),
34
35    /// The directory was created by a newer, unsupported disk format.
36    #[error("unsupported disk format {found}; this binary supports {supported}")]
37    UnsupportedFormat {
38        /// Version found in `FORMAT`.
39        found: u16,
40        /// Highest version understood by this binary.
41        supported: u16,
42    },
43
44    /// The immutable storage manifest could not be loaded or initialized.
45    #[error(transparent)]
46    Manifest(#[from] ManifestError),
47
48    /// A filesystem operation failed.
49    #[error("failed to {action} {path}: {source}")]
50    Io {
51        /// Operation that failed.
52        action: &'static str,
53        /// Path involved in the operation.
54        path: PathBuf,
55        /// Underlying operating-system error.
56        #[source]
57        source: io::Error,
58    },
59}
60
61impl From<StorageLimitError> for DataDirectoryError {
62    fn from(source: StorageLimitError) -> Self {
63        Self::Manifest(ManifestError::Io(limit_io_error(source)))
64    }
65}
66
67/// An exclusively owned Hyphae data directory.
68///
69/// The operating-system lock is held until this value is dropped. Opening the
70/// same directory for a second writer fails instead of relying on cooperative
71/// process behavior.
72#[derive(Debug)]
73pub struct DataDirectory {
74    root: PathBuf,
75    lock: File,
76    manifest: StorageManifest,
77    disk_format_version: u16,
78    #[cfg(test)]
79    fail_next_manifest_commit_after_write: bool,
80}
81
82impl DataDirectory {
83    /// Opens an existing data directory or initializes a new one.
84    ///
85    /// # Errors
86    ///
87    /// Returns an error when the directory cannot be initialized, is owned by
88    /// another writer, has a malformed marker, or uses a future format.
89    pub fn open(path: impl AsRef<Path>) -> Result<Self, DataDirectoryError> {
90        let limits = RecoveryLimits::compatibility();
91        let deadline = OperationDeadline::new(limits.timeout);
92        Self::open_with_limits_and_deadline(path, &limits, &deadline)
93    }
94
95    /// Opens or initializes a data directory under finite recovery limits.
96    ///
97    /// # Errors
98    ///
99    /// Returns an error for invalid limits, contention, malformed metadata,
100    /// future formats, exhausted directory policy, timeout, or I/O failure.
101    pub fn open_with_limits(
102        path: impl AsRef<Path>,
103        limits: &RecoveryLimits,
104    ) -> Result<Self, DataDirectoryError> {
105        limits.validate()?;
106        let deadline = OperationDeadline::new(limits.timeout);
107        Self::open_with_limits_and_deadline(path, limits, &deadline)
108    }
109
110    pub(crate) fn open_with_limits_and_deadline(
111        path: impl AsRef<Path>,
112        limits: &RecoveryLimits,
113        deadline: &OperationDeadline,
114    ) -> Result<Self, DataDirectoryError> {
115        deadline.check()?;
116        let root = path.as_ref().to_path_buf();
117        fs::create_dir_all(&root).map_err(|source| DataDirectoryError::Io {
118            action: "create data directory",
119            path: root.clone(),
120            source,
121        })?;
122
123        let lock_path = root.join("LOCK");
124        let lock = OpenOptions::new()
125            .create(true)
126            .read(true)
127            .write(true)
128            .truncate(false)
129            .open(&lock_path)
130            .map_err(|source| DataDirectoryError::Io {
131                action: "open lock file",
132                path: lock_path.clone(),
133                source,
134            })?;
135
136        match FileExt::try_lock(&lock) {
137            Ok(()) => {}
138            Err(TryLockError::WouldBlock) => {
139                return Err(DataDirectoryError::AlreadyLocked(root));
140            }
141            Err(TryLockError::Error(source)) => {
142                return Err(DataDirectoryError::Io {
143                    action: "lock data directory",
144                    path: lock_path,
145                    source,
146                });
147            }
148        }
149
150        let opened_format = initialize_or_validate_format(&root)?;
151        for name in REQUIRED_DIRECTORIES {
152            let directory = root.join(name);
153            fs::create_dir_all(&directory).map_err(|source| DataDirectoryError::Io {
154                action: "create data subdirectory",
155                path: directory,
156                source,
157            })?;
158        }
159
160        let manifest = StorageManifest::load_or_initialize_with_limits(
161            &root,
162            limits.max_directory_entries,
163            deadline,
164        )?;
165
166        Ok(Self {
167            root,
168            lock,
169            manifest,
170            disk_format_version: opened_format,
171            #[cfg(test)]
172            fail_next_manifest_commit_after_write: false,
173        })
174    }
175
176    /// Returns the canonical root path.
177    pub fn path(&self) -> &Path {
178        &self.root
179    }
180
181    /// Returns the format currently committed by the directory marker.
182    pub fn disk_format_version(&self) -> u16 {
183        self.disk_format_version
184    }
185
186    /// Returns the path of the active append-only log segment.
187    pub fn active_log_path(&self) -> PathBuf {
188        self.log_path(self.manifest.active_segment)
189    }
190
191    /// Opens the initial durable log while borrowing this directory lock.
192    ///
193    /// The returned writer cannot outlive the exclusive data-directory owner.
194    ///
195    /// # Errors
196    ///
197    /// Returns an error for I/O failures or any complete invalid frame.
198    pub fn open_log(&self) -> Result<OpenedLog<'_>, LogError> {
199        let limits = RecoveryLimits::compatibility();
200        let deadline = OperationDeadline::new(limits.timeout);
201        self.open_log_with_limits(&limits, &deadline)
202    }
203
204    pub(crate) fn open_log_with_limits(
205        &self,
206        limits: &RecoveryLimits,
207        deadline: &OperationDeadline,
208    ) -> Result<OpenedLog<'_>, LogError> {
209        let (log, recovery) = DurableLog::open_file_at_version_with_limits(
210            self.active_log_path(),
211            self.manifest.base_sequence,
212            self.manifest.base_digest,
213            self.disk_format_version,
214            limits,
215            deadline,
216        )?;
217        Ok(OpenedLog::new(log, recovery))
218    }
219
220    pub(crate) fn log_anchor(&self) -> (u64, [u8; 32]) {
221        (self.manifest.base_sequence, self.manifest.base_digest)
222    }
223
224    pub(crate) fn manifest(&self) -> StorageManifest {
225        self.manifest
226    }
227
228    pub(crate) fn log_path(&self, segment: u64) -> PathBuf {
229        self.root.join("log").join(format!("{segment:020}.hylog"))
230    }
231
232    pub(crate) fn snapshot_path(&self, sequence: u64) -> PathBuf {
233        self.root
234            .join("snapshots")
235            .join(format!("snapshot-{sequence:020}.hysnap"))
236    }
237
238    pub(crate) fn commit_manifest(
239        &mut self,
240        manifest: StorageManifest,
241    ) -> Result<(), DataDirectoryError> {
242        let limits = RecoveryLimits::default();
243        let deadline = OperationDeadline::new(limits.timeout);
244        self.commit_manifest_with_limits(manifest, &limits, &deadline)
245    }
246
247    pub(crate) fn commit_manifest_with_limits(
248        &mut self,
249        manifest: StorageManifest,
250        limits: &RecoveryLimits,
251        deadline: &OperationDeadline,
252    ) -> Result<(), DataDirectoryError> {
253        let target = manifest.path(&self.root);
254        let target_is_new = self.reserve_target_entry("manifest", &target, limits, deadline)?;
255        if target_is_new {
256            self.reserve_directory_entries("tmp", 1, limits, deadline)?;
257        }
258        manifest.write_new(&self.root)?;
259        #[cfg(test)]
260        if std::mem::take(&mut self.fail_next_manifest_commit_after_write) {
261            return Err(DataDirectoryError::Io {
262                action: "complete injected manifest commit",
263                path: target,
264                source: io::Error::other("injected post-write manifest failure"),
265            });
266        }
267        self.manifest = manifest;
268        Ok(())
269    }
270
271    #[cfg(test)]
272    pub(crate) fn inject_manifest_commit_failure_after_write(&mut self) {
273        self.fail_next_manifest_commit_after_write = true;
274    }
275
276    pub(crate) fn reserve_target_entry(
277        &self,
278        directory_name: &'static str,
279        target: &Path,
280        limits: &RecoveryLimits,
281        deadline: &OperationDeadline,
282    ) -> Result<bool, DataDirectoryError> {
283        deadline.check()?;
284        let target_is_new = !target
285            .try_exists()
286            .map_err(|source| DataDirectoryError::Io {
287                action: "inspect prospective storage entry",
288                path: target.to_path_buf(),
289                source,
290            })?;
291        self.reserve_directory_entries(directory_name, u64::from(target_is_new), limits, deadline)?;
292        Ok(target_is_new)
293    }
294
295    pub(crate) fn reserve_directory_entries(
296        &self,
297        directory_name: &'static str,
298        additional_entries: u64,
299        limits: &RecoveryLimits,
300        deadline: &OperationDeadline,
301    ) -> Result<(), DataDirectoryError> {
302        deadline.check()?;
303        let directory = self.root.join(directory_name);
304        let entries = fs::read_dir(&directory).map_err(|source| DataDirectoryError::Io {
305            action: "inspect storage directory capacity",
306            path: directory.clone(),
307            source,
308        })?;
309        let mut entry_count = 0_u64;
310        for entry in entries {
311            deadline.check()?;
312            entry.map_err(|source| DataDirectoryError::Io {
313                action: "inspect storage directory entry",
314                path: directory.clone(),
315                source,
316            })?;
317            entry_count =
318                entry_count
319                    .checked_add(1)
320                    .ok_or(StorageLimitError::DirectoryEntriesExceeded {
321                        maximum: limits.max_directory_entries,
322                    })?;
323            if entry_count > limits.max_directory_entries {
324                return Err(StorageLimitError::DirectoryEntriesExceeded {
325                    maximum: limits.max_directory_entries,
326                }
327                .into());
328            }
329        }
330        let required = entry_count.checked_add(additional_entries).ok_or(
331            StorageLimitError::DirectoryEntriesExceeded {
332                maximum: limits.max_directory_entries,
333            },
334        )?;
335        if required > limits.max_directory_entries {
336            return Err(StorageLimitError::DirectoryEntriesExceeded {
337                maximum: limits.max_directory_entries,
338            }
339            .into());
340        }
341        Ok(())
342    }
343
344    pub(crate) fn promote_format(&mut self) -> Result<(), DataDirectoryError> {
345        if self.disk_format_version == DISK_FORMAT_VERSION {
346            return Ok(());
347        }
348        write_format_marker(&self.root, DISK_FORMAT_VERSION)?;
349        self.disk_format_version = DISK_FORMAT_VERSION;
350        Ok(())
351    }
352
353    pub(crate) fn cleanup_retired_logs_with_limits(
354        &self,
355        limits: &RecoveryLimits,
356        deadline: &OperationDeadline,
357    ) -> bool {
358        let log_directory = self.root.join("log");
359        let Ok(entries) = fs::read_dir(&log_directory) else {
360            return false;
361        };
362        #[cfg(unix)]
363        let mut removed = false;
364        let mut entry_count = 0_u64;
365        for entry in entries.flatten() {
366            if deadline.check().is_err() {
367                return false;
368            }
369            entry_count = match entry_count.checked_add(1) {
370                Some(count) if count <= limits.max_directory_entries => count,
371                _ => return false,
372            };
373            let path = entry.path();
374            let Some(segment) = segment_from_path(&path) else {
375                continue;
376            };
377            if segment < self.manifest.active_segment {
378                match fs::remove_file(&path) {
379                    Ok(()) => {
380                        #[cfg(unix)]
381                        {
382                            removed = true;
383                        }
384                    }
385                    Err(source) if source.kind() == io::ErrorKind::NotFound => {}
386                    Err(_) => return false,
387                }
388            }
389        }
390        #[cfg(unix)]
391        if removed && sync_directory(&log_directory).is_err() {
392            return false;
393        }
394        true
395    }
396}
397
398impl Drop for DataDirectory {
399    fn drop(&mut self) {
400        let _ignored = FileExt::unlock(&self.lock);
401    }
402}
403
404fn initialize_or_validate_format(root: &Path) -> Result<u16, DataDirectoryError> {
405    let format_path = root.join("FORMAT");
406    if format_path.exists() {
407        return validate_format(&format_path);
408    }
409    write_format_marker(root, DISK_FORMAT_VERSION)?;
410    Ok(DISK_FORMAT_VERSION)
411}
412
413fn write_format_marker(root: &Path, version: u16) -> Result<(), DataDirectoryError> {
414    let format_path = root.join("FORMAT");
415    let temporary_path = root.join("FORMAT.new");
416    let mut file = OpenOptions::new()
417        .create(true)
418        .write(true)
419        .truncate(true)
420        .open(&temporary_path)
421        .map_err(|source| DataDirectoryError::Io {
422            action: "create temporary format marker",
423            path: temporary_path.clone(),
424            source,
425        })?;
426    let marker = format!("{FORMAT_PREFIX}{version}\n");
427    file.write_all(marker.as_bytes())
428        .and_then(|()| file.sync_all())
429        .map_err(|source| DataDirectoryError::Io {
430            action: "initialize temporary format marker",
431            path: temporary_path.clone(),
432            source,
433        })?;
434    drop(file);
435    fs::rename(&temporary_path, &format_path).map_err(|source| DataDirectoryError::Io {
436        action: "promote format marker",
437        path: format_path,
438        source,
439    })?;
440    #[cfg(unix)]
441    sync_directory(root)?;
442    Ok(())
443}
444
445#[cfg(unix)]
446fn sync_directory(path: &Path) -> Result<(), DataDirectoryError> {
447    File::open(path)
448        .and_then(|directory| directory.sync_all())
449        .map_err(|source| DataDirectoryError::Io {
450            action: "synchronize data directory",
451            path: path.to_path_buf(),
452            source,
453        })
454}
455
456fn validate_format(path: &Path) -> Result<u16, DataDirectoryError> {
457    let file = File::open(path).map_err(|source| DataDirectoryError::Io {
458        action: "open format marker",
459        path: path.to_path_buf(),
460        source,
461    })?;
462    let mut marker_bytes = Vec::with_capacity(MAX_FORMAT_MARKER_BYTES + 1);
463    file.take(FORMAT_MARKER_READ_LIMIT)
464        .read_to_end(&mut marker_bytes)
465        .map_err(|source| DataDirectoryError::Io {
466            action: "read format marker",
467            path: path.to_path_buf(),
468            source,
469        })?;
470    if marker_bytes.len() > MAX_FORMAT_MARKER_BYTES {
471        return Err(DataDirectoryError::MalformedFormat(path.to_path_buf()));
472    }
473    let marker = String::from_utf8(marker_bytes)
474        .map_err(|_| DataDirectoryError::MalformedFormat(path.to_path_buf()))?;
475
476    let Some(raw_version) = marker
477        .strip_prefix(FORMAT_PREFIX)
478        .and_then(|value| value.strip_suffix('\n'))
479    else {
480        return Err(DataDirectoryError::MalformedFormat(path.to_path_buf()));
481    };
482    let version = raw_version
483        .parse::<u16>()
484        .map_err(|_| DataDirectoryError::MalformedFormat(path.to_path_buf()))?;
485    if version > DISK_FORMAT_VERSION {
486        return Err(DataDirectoryError::UnsupportedFormat {
487            found: version,
488            supported: DISK_FORMAT_VERSION,
489        });
490    }
491    if version < MIN_DISK_FORMAT_VERSION {
492        return Err(DataDirectoryError::MalformedFormat(path.to_path_buf()));
493    }
494    Ok(version)
495}
496
497fn segment_from_path(path: &Path) -> Option<u64> {
498    let filename = path.file_name()?.to_str()?;
499    let raw_segment = filename.strip_suffix(".hylog")?;
500    if raw_segment.len() != 20 || !raw_segment.bytes().all(|byte| byte.is_ascii_digit()) {
501        return None;
502    }
503    let segment = raw_segment.parse().ok()?;
504    (format!("{segment:020}.hylog") == filename).then_some(segment)
505}
506
507#[cfg(test)]
508mod tests {
509    use std::{error::Error, fs};
510
511    use uuid::Uuid;
512
513    use super::{DataDirectory, DataDirectoryError, REQUIRED_DIRECTORIES};
514    use crate::{DurableLog, test_support::TestDirectory};
515
516    #[test]
517    fn initializes_canonical_layout() -> Result<(), Box<dyn Error>> {
518        let temporary = TestDirectory::new("data-layout")?;
519        let root = temporary.path().join("data");
520        let directory = DataDirectory::open(&root)?;
521
522        assert_eq!(directory.path(), root);
523        assert_eq!(
524            fs::read_to_string(root.join("FORMAT"))?,
525            "hyphae-disk-format=2\n"
526        );
527        assert!(root.join("LOCK").is_file());
528        for name in REQUIRED_DIRECTORIES {
529            assert!(root.join(name).is_dir());
530        }
531        Ok(())
532    }
533
534    #[test]
535    fn rejects_a_second_writer() -> Result<(), Box<dyn Error>> {
536        let temporary = TestDirectory::new("data-lock")?;
537        let first = DataDirectory::open(temporary.path())?;
538        let second = DataDirectory::open(temporary.path());
539
540        assert!(matches!(second, Err(DataDirectoryError::AlreadyLocked(_))));
541        drop(first);
542        DataDirectory::open(temporary.path())?;
543        Ok(())
544    }
545
546    #[test]
547    fn rejects_future_format() -> Result<(), Box<dyn Error>> {
548        let temporary = TestDirectory::new("future-format")?;
549        fs::write(temporary.path().join("FORMAT"), "hyphae-disk-format=3\n")?;
550
551        let result = DataDirectory::open(temporary.path());
552        assert!(matches!(
553            result,
554            Err(DataDirectoryError::UnsupportedFormat {
555                found: 3,
556                supported: 2
557            })
558        ));
559        Ok(())
560    }
561
562    #[test]
563    fn format_marker_reads_are_bounded_at_sixty_four_bytes() -> Result<(), Box<dyn Error>> {
564        for length in [64_usize, 65] {
565            let temporary = TestDirectory::new(&format!("bounded-format-{length}"))?;
566            fs::write(temporary.path().join("FORMAT"), vec![b'x'; length])?;
567            assert!(matches!(
568                DataDirectory::open(temporary.path()),
569                Err(DataDirectoryError::MalformedFormat(_))
570            ));
571        }
572        Ok(())
573    }
574
575    #[test]
576    fn durable_log_cannot_outlive_directory_lock() -> Result<(), Box<dyn Error>> {
577        let temporary = TestDirectory::new("locked-log")?;
578        let directory = DataDirectory::open(temporary.path())?;
579        let mut opened = directory.open_log()?;
580        opened
581            .log
582            .append_transaction(Uuid::now_v7(), &[b"operation".to_vec()])?;
583
584        assert_eq!(opened.recovery.transactions.len(), 0);
585        assert!(matches!(
586            DataDirectory::open(temporary.path()),
587            Err(DataDirectoryError::AlreadyLocked(_))
588        ));
589        Ok(())
590    }
591
592    #[test]
593    fn migrates_an_existing_format_one_directory_without_a_manifest() -> Result<(), Box<dyn Error>>
594    {
595        let temporary = TestDirectory::new("data-manifest-migration")?;
596        let root = temporary.path().join("data");
597        fs::create_dir_all(root.join("log"))?;
598        fs::write(root.join("FORMAT"), "hyphae-disk-format=1\n")?;
599        let log_path = root.join("log/00000000000000000001.hylog");
600        let (mut legacy_log, _) = DurableLog::open_file_at_version(&log_path, 0, [0; 32], 1)?;
601        legacy_log.append_transaction(Uuid::now_v7(), &[b"preserved".to_vec()])?;
602        drop(legacy_log);
603
604        let directory = DataDirectory::open(&root)?;
605        assert!(
606            root.join("manifest/00000000000000000001.hymanifest")
607                .is_file()
608        );
609        let opened = directory.open_log()?;
610        assert_eq!(opened.recovery.transactions.len(), 1);
611        assert_eq!(opened.recovery.transactions[0].operations[0], b"preserved");
612        Ok(())
613    }
614}
hyphae_storage/data_directory.rs

hyphae_storage/
data_directory.rs
