core_storage/
fs.rs

1use std::fs::{File, OpenOptions};
2use std::io::{Read, Write};
3use std::path::PathBuf;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
6pub enum FileId {
7    Wal,
8    Snapshot,
9}
10
11impl FileId {
12    fn name(self) -> &'static str {
13        match self {
14            FileId::Wal => "wal.bin",
15            FileId::Snapshot => "snapshot.bin",
16        }
17    }
18}
19
20pub trait Fs {
21    fn append(&mut self, file: FileId, data: &[u8]) -> std::io::Result<()>;
22    fn sync(&mut self, file: FileId) -> std::io::Result<()>;
23    fn read(&self, file: FileId) -> std::io::Result<Vec<u8>>;
24    fn write_atomic(&mut self, file: FileId, data: &[u8]) -> std::io::Result<()>;
25}
26
27pub trait FsIntrospect {
28    fn total_appended(&self) -> usize;
29    fn sync_count(&self) -> usize {
30        0
31    }
32}
33
34#[derive(Debug)]
35pub struct RealFs {
36    dir: PathBuf,
37}
38
39impl RealFs {
40    pub fn new(dir: &std::path::Path) -> std::io::Result<Self> {
41        std::fs::create_dir_all(dir)?;
42        Ok(Self {
43            dir: dir.to_path_buf(),
44        })
45    }
46
47    fn path(&self, file: FileId) -> PathBuf {
48        self.dir.join(file.name())
49    }
50}
51
52impl Fs for RealFs {
53    fn append(&mut self, file: FileId, data: &[u8]) -> std::io::Result<()> {
54        let mut f = OpenOptions::new()
55            .create(true)
56            .append(true)
57            .open(self.path(file))?;
58        f.write_all(data)
59    }
60
61    fn sync(&mut self, file: FileId) -> std::io::Result<()> {
62        let f = File::open(self.path(file))?;
63        full_sync(&f)
64    }
65
66    fn read(&self, file: FileId) -> std::io::Result<Vec<u8>> {
67        match File::open(self.path(file)) {
68            Ok(mut f) => {
69                let mut buf = Vec::new();
70                f.read_to_end(&mut buf)?;
71                Ok(buf)
72            }
73            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
74            Err(e) => Err(e),
75        }
76    }
77
78    fn write_atomic(&mut self, file: FileId, data: &[u8]) -> std::io::Result<()> {
79        let tmp = self.dir.join(format!("{}.tmp", file.name()));
80        {
81            let mut f = File::create(&tmp)?;
82            f.write_all(data)?;
83            full_sync(&f)?;
84        }
85        std::fs::rename(&tmp, self.path(file))?;
86        sync_dir(&self.dir)
87    }
88}
89
90fn full_sync(file: &File) -> std::io::Result<()> {
91    #[cfg(target_os = "macos")]
92    {
93        use std::os::unix::io::AsRawFd;
94        let fd = file.as_raw_fd();
95        let rc = unsafe { libc::fcntl(fd, libc::F_FULLFSYNC) };
96        if rc == -1 {
97            return Err(std::io::Error::last_os_error());
98        }
99        Ok(())
100    }
101    #[cfg(not(target_os = "macos"))]
102    {
103        file.sync_all()
104    }
105}
106
107fn sync_dir(dir: &std::path::Path) -> std::io::Result<()> {
108    let d = File::open(dir)?;
109    d.sync_all()
110}
111
112#[cfg(test)]
113mod tests {
114    use super::*;
115
116    fn tmp() -> std::path::PathBuf {
117        let d = std::env::temp_dir().join(format!("graphdb-fs-{}", std::process::id()));
118        let _ = std::fs::remove_dir_all(&d);
119        d
120    }
121
122    #[test]
123    fn append_read_and_atomic_write() {
124        let mut fs = RealFs::new(&tmp()).unwrap();
125        assert_eq!(fs.read(FileId::Wal).unwrap(), Vec::<u8>::new()); // absent = empty
126        fs.append(FileId::Wal, b"ab").unwrap();
127        fs.append(FileId::Wal, b"cd").unwrap();
128        fs.sync(FileId::Wal).unwrap();
129        assert_eq!(fs.read(FileId::Wal).unwrap(), b"abcd");
130        fs.write_atomic(FileId::Snapshot, b"snap1").unwrap();
131        fs.write_atomic(FileId::Snapshot, b"snap2").unwrap(); // replaces
132        assert_eq!(fs.read(FileId::Snapshot).unwrap(), b"snap2");
133        fs.write_atomic(FileId::Wal, b"").unwrap(); // truncation path
134        assert_eq!(fs.read(FileId::Wal).unwrap(), Vec::<u8>::new());
135    }
136
137    #[test]
138    fn write_atomic_replaces_and_still_readable() {
139        // existing append_read_and_atomic_write already covers replace;
140        // keep it; dir-sync is best-effort observable only via crash tests.
141        // Do not fake F_FULLFSYNC in SimFs.
142        let d = std::env::temp_dir().join(format!("graphdb-fs-atomic-{}", std::process::id()));
143        let _ = std::fs::remove_dir_all(&d);
144        let mut fs = RealFs::new(&d).unwrap();
145        fs.write_atomic(FileId::Snapshot, b"snap1").unwrap();
146        fs.write_atomic(FileId::Snapshot, b"snap2").unwrap();
147        assert_eq!(fs.read(FileId::Snapshot).unwrap(), b"snap2");
148    }
149}
core_storage/fs.rs

core_storage/
fs.rs
