hashtree_lmdb/
physical_space.rs

1//! Opt-in admission at actual local write boundaries, independent of blob size.
2//!
3//! The full filesystem-block-rounded write range is charged, even for writes
4//! that reuse allocated pages. A 16 MiB margin covers ordinary allocation and
5//! namespace metadata beyond that data range. This is a fail-closed operational
6//! guard, not a filesystem quota: unrelated processes and filesystem-specific
7//! allocation internals cannot be reserved by statvfs. No guarded data syscall
8//! exceeds 64 MiB; external files are admitted in at most 1 MiB chunks.
9
10use std::fs::File;
11use std::io::{self, Seek, Write};
12use std::path::Path;
13use std::sync::atomic::{AtomicI32, Ordering};
14use std::sync::Arc;
15
16pub const PHYSICAL_SPACE_METADATA_MARGIN: u64 = 16 * 1024 * 1024;
17pub const MAX_GUARDED_WRITE_BYTES: usize = 64 * 1024 * 1024;
18const EXTERNAL_WRITE_BYTES: usize = 1024 * 1024;
19
20#[derive(Clone)]
21pub struct PhysicalSpaceGuard {
22    inner: Arc<GuardInner>,
23    #[cfg(unix)]
24    admission: heed::WriteAdmission,
25}
26
27struct GuardInner {
28    minimum_free_bytes: u64,
29    refusal: AtomicI32,
30}
31
32impl std::fmt::Debug for PhysicalSpaceGuard {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        f.debug_struct("PhysicalSpaceGuard")
35            .field("minimum_free_bytes", &self.inner.minimum_free_bytes)
36            .finish()
37    }
38}
39
40impl PhysicalSpaceGuard {
41    pub fn new(minimum_free_bytes: u64) -> io::Result<Self> {
42        #[cfg(unix)]
43        {
44            let inner = Arc::new(GuardInner {
45                minimum_free_bytes,
46                refusal: AtomicI32::new(0),
47            });
48            let callback = Arc::clone(&inner);
49            let admission = heed::WriteAdmission::new(move |fd, offset, length| {
50                callback.admit(fd, offset, length)
51            });
52            Ok(Self { inner, admission })
53        }
54        #[cfg(not(unix))]
55        {
56            let _ = minimum_free_bytes;
57            Err(io::Error::new(
58                io::ErrorKind::Unsupported,
59                "physical write admission requires Unix descriptors",
60            ))
61        }
62    }
63
64    /// Any admission failure ends this guarded writer; a fresh process may retry.
65    pub fn has_refused(&self) -> bool {
66        self.inner.refusal.load(Ordering::Acquire) != 0
67    }
68
69    pub fn refusal_error(&self) -> io::Error {
70        io::Error::from_raw_os_error(self.inner.refusal.load(Ordering::Acquire).max(1))
71    }
72
73    pub(crate) fn stop_after_write_error(&self, error: &hashtree_core::store::StoreError) -> bool {
74        if let hashtree_core::store::StoreError::Io(error) = error {
75            self.inner
76                .refusal
77                .compare_exchange(
78                    0,
79                    error.raw_os_error().unwrap_or(libc::EIO),
80                    Ordering::AcqRel,
81                    Ordering::Acquire,
82                )
83                .ok();
84        }
85        self.has_refused()
86    }
87
88    pub fn status(&self) -> String {
89        format!("physical-space admission: floor={} metadata_margin={} max_write_quantum={} latched_errno={}",
90            self.inner.minimum_free_bytes, PHYSICAL_SPACE_METADATA_MARGIN, MAX_GUARDED_WRITE_BYTES,
91            self.inner.refusal.load(Ordering::Acquire))
92    }
93
94    pub fn install(&self, options: &mut heed::EnvOpenOptions) -> io::Result<()> {
95        #[cfg(unix)]
96        {
97            options.write_admission(self.admission.clone());
98            Ok(())
99        }
100        #[cfg(not(unix))]
101        {
102            let _ = options;
103            Err(io::Error::new(
104                io::ErrorKind::Unsupported,
105                "physical write admission requires Unix descriptors",
106            ))
107        }
108    }
109
110    /// Admit each namespace change on its actual parent filesystem. Existing
111    /// directories require no admission, preserving duplicate/no-op operation.
112    pub fn create_dir_all(&self, path: &Path) -> io::Result<()> {
113        if path.is_dir() {
114            return Ok(());
115        }
116        let parent = path
117            .parent()
118            .filter(|p| !p.as_os_str().is_empty())
119            .unwrap_or(Path::new("."));
120        self.create_dir_all(parent)?;
121        self.admit_file(&File::open(parent)?, 0, 1)?;
122        match std::fs::create_dir(path) {
123            Ok(()) => Ok(()),
124            Err(error) if error.kind() == io::ErrorKind::AlreadyExists && path.is_dir() => Ok(()),
125            Err(error) => Err(error),
126        }
127    }
128
129    pub fn admit_file(&self, file: &File, offset: u64, length: usize) -> io::Result<()> {
130        #[cfg(unix)]
131        {
132            use std::os::fd::AsRawFd;
133            self.inner.admit(file.as_raw_fd(), offset, length)
134        }
135        #[cfg(not(unix))]
136        {
137            let _ = (file, offset, length);
138            Err(io::Error::new(
139                io::ErrorKind::Unsupported,
140                "physical write admission requires Unix descriptors",
141            ))
142        }
143    }
144
145    pub fn write_all(&self, file: &mut File, data: &[u8]) -> io::Result<()> {
146        for chunk in data.chunks(EXTERNAL_WRITE_BYTES) {
147            let offset = file.stream_position()?;
148            self.admit_file(file, offset, chunk.len())?;
149            file.write_all(chunk)?;
150        }
151        Ok(())
152    }
153}
154
155impl GuardInner {
156    #[cfg(unix)]
157    fn admit(&self, fd: std::os::fd::RawFd, offset: u64, length: usize) -> io::Result<()> {
158        let refused = self.refusal.load(Ordering::Acquire);
159        if refused != 0 {
160            return Err(io::Error::from_raw_os_error(refused));
161        }
162        let result = self.admit_checked(fd, offset, length);
163        if let Err(error) = &result {
164            self.refusal
165                .compare_exchange(
166                    0,
167                    error.raw_os_error().unwrap_or(libc::EIO),
168                    Ordering::AcqRel,
169                    Ordering::Acquire,
170                )
171                .ok();
172        }
173        result
174    }
175
176    #[cfg(unix)]
177    fn admit_checked(&self, fd: std::os::fd::RawFd, offset: u64, length: usize) -> io::Result<()> {
178        if length == 0 {
179            return Ok(());
180        }
181        if length > MAX_GUARDED_WRITE_BYTES {
182            return Err(io::Error::from_raw_os_error(libc::EFBIG));
183        }
184        let mut stats = std::mem::MaybeUninit::<libc::statvfs>::uninit();
185        // fd is owned by the active LMDB/file write and remains valid here.
186        if unsafe { libc::fstatvfs(fd, stats.as_mut_ptr()) } != 0 {
187            return Err(io::Error::last_os_error());
188        }
189        let stats = unsafe { stats.assume_init() };
190        let block = stats.f_frsize as u64;
191        if block == 0 {
192            return Err(io::Error::from_raw_os_error(libc::EIO));
193        }
194        let available = (stats.f_bavail as u64).saturating_mul(block);
195        let touched = (offset % block)
196            .checked_add(length as u64)
197            .and_then(|n| n.checked_add(block - 1))
198            .map(|n| n / block * block)
199            .ok_or_else(|| io::Error::from_raw_os_error(libc::EFBIG))?;
200        let required = self
201            .minimum_free_bytes
202            .checked_add(PHYSICAL_SPACE_METADATA_MARGIN)
203            .and_then(|n| n.checked_add(touched))
204            .ok_or_else(|| io::Error::from_raw_os_error(libc::ENOSPC))?;
205        if available < required {
206            return Err(io::Error::from_raw_os_error(libc::ENOSPC));
207        }
208        Ok(())
209    }
210}
211
212#[cfg(all(test, unix))]
213mod tests {
214    use super::*;
215    use std::os::fd::AsRawFd;
216
217    #[test]
218    fn failed_filesystem_query_latches_and_cannot_be_retried_as_a_write() {
219        let guard = PhysicalSpaceGuard::new(0).unwrap();
220        let error = guard.inner.admit(-1, 0, 1).unwrap_err();
221        assert_eq!(error.raw_os_error(), Some(libc::EBADF));
222        let temp = tempfile::tempfile().unwrap();
223        assert_eq!(
224            guard
225                .inner
226                .admit(temp.as_raw_fd(), 0, 1)
227                .unwrap_err()
228                .raw_os_error(),
229            Some(libc::EBADF)
230        );
231        assert!(guard.has_refused());
232        assert!(guard.status().contains("latched_errno=9"));
233    }
234
235    #[test]
236    fn unsupported_write_quantum_is_rejected_before_any_write() {
237        let guard = PhysicalSpaceGuard::new(0).unwrap();
238        let file = tempfile::tempfile().unwrap();
239        let error = guard
240            .admit_file(&file, 0, MAX_GUARDED_WRITE_BYTES + 1)
241            .unwrap_err();
242        assert_eq!(error.raw_os_error(), Some(libc::EFBIG));
243        assert_eq!(file.metadata().unwrap().len(), 0);
244    }
245}
hashtree_lmdb/physical_space.rs

hashtree_lmdb/
physical_space.rs
