Skip to main content

zeppelin_embed/vfs/
mod.rs

1//! Synchronous virtual-filesystem seam for crash and fault injection.
2
3#[cfg(any(test, feature = "test-seams"))]
4pub mod crash;
5#[cfg(any(test, feature = "test-seams"))]
6pub mod fault;
7
8use std::fs::{File, OpenOptions};
9use std::io::{IoSlice, Read, Seek, SeekFrom, Write};
10#[cfg(unix)]
11use std::os::fd::AsRawFd;
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14use std::sync::atomic::{AtomicU64, Ordering};
15
16/// Storage-ordering primitive requested by a persisted commit.
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18pub enum SyncKind {
19    /// Order prior writes before later writes without requesting durable media.
20    ///
21    /// Linux has no barrier-only primitive, so `ordered` durability degrades
22    /// to `fdatasync` on Linux.
23    Barrier,
24    /// Flush prior writes through the platform's durable-media primitive.
25    Full,
26}
27
28/// One open append-only file owned by a WAL writer.
29pub trait VfsFile: Send {
30    /// Appends every supplied byte without reopening the path.
31    fn append(&mut self, bytes: &[u8]) -> std::io::Result<()>;
32    /// Appends one logical group from discontiguous owned buffers.
33    ///
34    /// The default preserves compatibility for test filesystems. Production
35    /// files override it with one vectored write sequence, so WAL visibility
36    /// and persistence can share encoded record storage without coalescing it.
37    fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
38        for buffer in buffers {
39            self.append(buffer)?;
40        }
41        Ok(())
42    }
43    /// Synchronizes this already-open handle using the named primitive.
44    fn sync(&self, kind: SyncKind) -> std::io::Result<()>;
45}
46
47/// Minimal synchronous filesystem operations used by persisted commits.
48pub trait Vfs: Send + Sync {
49    /// Supplies an optional counter for mmap-backed segment payload accesses.
50    #[doc(hidden)]
51    fn segment_data_read_counter(&self) -> Option<Arc<AtomicU64>> {
52        None
53    }
54    /// Admits a store directory, creating missing ancestors when requested,
55    /// and reports whether the resulting path is a directory.
56    fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool>;
57    /// Exclusively creates one directory. Existing paths are preserved and
58    /// reported as `AlreadyExists`; no ancestor is adopted implicitly.
59    fn create_directory(&self, _path: &Path) -> std::io::Result<()> {
60        Err(std::io::Error::new(
61            std::io::ErrorKind::Unsupported,
62            "VFS does not support exclusive directory creation",
63        ))
64    }
65    /// Removes one empty directory. A non-empty directory is an error.
66    fn remove_directory(&self, _path: &Path) -> std::io::Result<()> {
67        Err(std::io::Error::new(
68            std::io::ErrorKind::Unsupported,
69            "VFS does not support directory removal",
70        ))
71    }
72    /// Opens an existing path and returns its byte length.
73    fn open(&self, path: &Path) -> std::io::Result<u64>;
74    /// Opens an existing path as an owned file suitable for memory mapping.
75    ///
76    /// Test filesystems without a native file descriptor return
77    /// [`std::io::ErrorKind::Unsupported`]. Consequently `MemoryVfs`,
78    /// `CrashVfs`, and memory-backed `FaultVfs` images cannot publish or reopen
79    /// a sealed segment; use a path-backed VFS when exercising that lifecycle.
80    fn open_for_map(&self, path: &Path) -> std::io::Result<File>;
81    /// Reads an entire file.
82    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>>;
83    /// Reads at most `length` bytes beginning at `offset`.
84    fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>>;
85    /// Creates or truncates a file and writes all bytes.
86    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()>;
87    /// Removes a suffix without copying the retained file bytes.
88    fn truncate(&self, _path: &Path, _length: u64) -> std::io::Result<()> {
89        Err(std::io::Error::new(
90            std::io::ErrorKind::Unsupported,
91            "VFS does not support suffix truncation",
92        ))
93    }
94    /// Exclusively creates a file and writes every byte without replacing any
95    /// existing path. A failed write may leave a new partial file; callers must
96    /// retain it as an uncommitted artifact until safe cleanup is established.
97    /// Filesystems without atomic exclusive creation fail explicitly.
98    fn create_new(&self, _path: &Path, _bytes: &[u8]) -> std::io::Result<()> {
99        Err(std::io::Error::new(
100            std::io::ErrorKind::Unsupported,
101            "VFS does not support exclusive file creation",
102        ))
103    }
104    /// Opens or creates one file for handle-oriented append-only writes.
105    fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>>;
106    /// Atomically renames one path over another according to platform semantics.
107    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()>;
108    /// Synchronizes a file or directory using an explicit platform primitive.
109    fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()>;
110    /// Lists direct children of a directory.
111    fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>>;
112    /// Visits direct children one at a time without collecting the directory.
113    fn for_each_direct_child(
114        &self,
115        directory: &Path,
116        visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
117    ) -> std::io::Result<()>;
118    /// Deletes one file.
119    fn delete(&self, path: &Path) -> std::io::Result<()>;
120}
121
122/// Standard-library-backed production filesystem.
123#[derive(Clone, Copy, Debug, Default)]
124pub struct StdVfs;
125
126struct StdVfsFile(File);
127
128impl VfsFile for StdVfsFile {
129    fn append(&mut self, bytes: &[u8]) -> std::io::Result<()> {
130        #[cfg(feature = "test-seams")]
131        file_test_support::event(file_test_support::FileEvent::BeforeAppend)?;
132        self.0.write_all(bytes)
133    }
134
135    fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
136        let mut remaining = buffers;
137        while !remaining.is_empty() {
138            let written = self.0.write_vectored(remaining)?;
139            if written == 0 {
140                return Err(std::io::Error::new(
141                    std::io::ErrorKind::WriteZero,
142                    "failed to append vectored WAL group",
143                ));
144            }
145            IoSlice::advance_slices(&mut remaining, written);
146        }
147        Ok(())
148    }
149
150    fn sync(&self, kind: SyncKind) -> std::io::Result<()> {
151        #[cfg(feature = "test-seams")]
152        file_test_support::event(file_test_support::FileEvent::BeforeSync)?;
153        sync_file(&self.0, kind)?;
154        #[cfg(feature = "test-seams")]
155        file_test_support::event(file_test_support::FileEvent::AfterSync)?;
156        Ok(())
157    }
158}
159
160impl Vfs for StdVfs {
161    fn truncate(&self, path: &Path, length: u64) -> std::io::Result<()> {
162        OpenOptions::new().write(true).open(path)?.set_len(length)
163    }
164
165    fn create_new(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
166        let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
167        file.write_all(bytes)
168    }
169
170    fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool> {
171        if create {
172            std::fs::create_dir_all(path)?;
173        }
174        Ok(std::fs::metadata(path)?.is_dir())
175    }
176
177    fn create_directory(&self, path: &Path) -> std::io::Result<()> {
178        std::fs::create_dir(path)
179    }
180
181    fn remove_directory(&self, path: &Path) -> std::io::Result<()> {
182        std::fs::remove_dir(path)
183    }
184
185    fn open(&self, path: &Path) -> std::io::Result<u64> {
186        Ok(File::open(path)?.metadata()?.len())
187    }
188
189    fn open_for_map(&self, path: &Path) -> std::io::Result<File> {
190        File::open(path)
191    }
192
193    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
194        let mut file = File::open(path)?;
195        let length = usize::try_from(file.metadata()?.len()).map_err(|_| {
196            std::io::Error::new(std::io::ErrorKind::InvalidData, "file length exceeds usize")
197        })?;
198        let mut bytes = Vec::with_capacity(length);
199        file.read_to_end(&mut bytes)?;
200        Ok(bytes)
201    }
202
203    fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>> {
204        let mut file = File::open(path)?;
205        file.seek(SeekFrom::Start(offset))?;
206        let mut bytes = vec![0_u8; length];
207        let mut filled = 0_usize;
208        while filled < length {
209            let remaining = bytes.get_mut(filled..).ok_or_else(|| {
210                std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid read range")
211            })?;
212            let read = file.read(remaining)?;
213            if read == 0 {
214                break;
215            }
216            filled = filled.saturating_add(read);
217        }
218        bytes.truncate(filled);
219        Ok(bytes)
220    }
221
222    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
223        let mut file = OpenOptions::new()
224            .create(true)
225            .truncate(true)
226            .write(true)
227            .open(path)?;
228        file.write_all(bytes)
229    }
230
231    fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>> {
232        let file = OpenOptions::new().create(true).append(true).open(path)?;
233        Ok(Box::new(StdVfsFile(file)))
234    }
235
236    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
237        std::fs::rename(from, to)
238    }
239
240    fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()> {
241        sync_path(path, kind)
242    }
243
244    fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>> {
245        std::fs::read_dir(directory)?
246            .map(|entry| entry.map(|value| value.path()))
247            .collect()
248    }
249
250    fn for_each_direct_child(
251        &self,
252        directory: &Path,
253        visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
254    ) -> std::io::Result<()> {
255        for entry in std::fs::read_dir(directory)? {
256            let entry = entry?;
257            visitor(&entry.path())?;
258        }
259        Ok(())
260    }
261
262    fn delete(&self, path: &Path) -> std::io::Result<()> {
263        std::fs::remove_file(path)
264    }
265}
266
267/// Byte/call-counting decorator used by deterministic open-cost gates.
268pub struct CountingVfs<V> {
269    inner: V,
270    counters: Arc<CountingVfsCounters>,
271    read_path_filter: Option<PathBuf>,
272}
273
274#[derive(Default)]
275struct CountingVfsCounters {
276    open_calls: AtomicU64,
277    open_for_map_calls: AtomicU64,
278    read_calls: AtomicU64,
279    filtered_read_calls: AtomicU64,
280    read_bytes: AtomicU64,
281    segment_bytes_read: Arc<AtomicU64>,
282    write_calls: AtomicU64,
283    bytes_written: AtomicU64,
284    append_calls: AtomicU64,
285    bytes_appended: AtomicU64,
286    rename_calls: AtomicU64,
287    delete_calls: AtomicU64,
288    enumerate_calls: AtomicU64,
289    barrier_sync_calls: AtomicU64,
290    full_sync_calls: AtomicU64,
291    handle_barrier_sync_calls: AtomicU64,
292    handle_full_sync_calls: AtomicU64,
293}
294
295struct CountingVfsFile {
296    inner: Box<dyn VfsFile>,
297    counters: Arc<CountingVfsCounters>,
298}
299
300impl VfsFile for CountingVfsFile {
301    fn append(&mut self, bytes: &[u8]) -> std::io::Result<()> {
302        self.inner.append(bytes)?;
303        self.counters.append_calls.fetch_add(1, Ordering::Relaxed);
304        self.counters
305            .bytes_appended
306            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
307        Ok(())
308    }
309
310    fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
311        let bytes = buffers
312            .iter()
313            .fold(0_usize, |total, buffer| total.saturating_add(buffer.len()));
314        self.inner.append_vectored(buffers)?;
315        self.counters.append_calls.fetch_add(1, Ordering::Relaxed);
316        self.counters
317            .bytes_appended
318            .fetch_add(bytes as u64, Ordering::Relaxed);
319        Ok(())
320    }
321
322    fn sync(&self, kind: SyncKind) -> std::io::Result<()> {
323        self.inner.sync(kind)?;
324        match kind {
325            SyncKind::Barrier => &self.counters.handle_barrier_sync_calls,
326            SyncKind::Full => &self.counters.handle_full_sync_calls,
327        }
328        .fetch_add(1, Ordering::Relaxed);
329        Ok(())
330    }
331}
332
333impl<V> CountingVfs<V> {
334    /// Wraps a filesystem with zeroed counters.
335    #[must_use]
336    pub fn new(inner: V) -> Self {
337        Self {
338            inner,
339            counters: Arc::new(CountingVfsCounters::default()),
340            read_path_filter: None,
341        }
342    }
343
344    /// Counts successful reads of one exact path separately from global reads.
345    #[must_use]
346    pub fn with_read_path_filter(mut self, path: impl Into<PathBuf>) -> Self {
347        self.read_path_filter = Some(path.into());
348        self
349    }
350
351    /// Returns the wrapped filesystem.
352    #[must_use]
353    pub const fn inner(&self) -> &V {
354        &self.inner
355    }
356
357    /// Runs a test probe with mmap-backed segment payload accounting enabled.
358    #[doc(hidden)]
359    pub fn audit_segment_data_reads<T>(&self, operation: impl FnOnce() -> T) -> T {
360        let _audit = crate::segment::reader::install_data_read_audit(Some(Arc::clone(
361            &self.counters.segment_bytes_read,
362        )));
363        operation()
364    }
365
366    /// Returns observed open calls.
367    #[must_use]
368    pub fn open_calls(&self) -> u64 {
369        self.counters.open_calls.load(Ordering::Relaxed)
370    }
371
372    /// Returns observed mapping-file open calls.
373    #[must_use]
374    pub fn open_for_map_calls(&self) -> u64 {
375        self.counters.open_for_map_calls.load(Ordering::Relaxed)
376    }
377
378    /// Returns observed read calls.
379    #[must_use]
380    pub fn read_calls(&self) -> u64 {
381        self.counters.read_calls.load(Ordering::Relaxed)
382    }
383
384    /// Returns successful reads matching the configured exact path.
385    #[must_use]
386    pub fn filtered_read_calls(&self) -> u64 {
387        self.counters.filtered_read_calls.load(Ordering::Relaxed)
388    }
389
390    /// Returns exact bytes returned by reads.
391    #[must_use]
392    pub fn read_bytes(&self) -> u64 {
393        self.counters.read_bytes.load(Ordering::Relaxed)
394    }
395
396    /// Returns exact bytes read from immutable segment paths.
397    #[must_use]
398    pub fn segment_bytes_read(&self) -> u64 {
399        self.counters.segment_bytes_read.load(Ordering::Relaxed)
400    }
401
402    /// Returns observed whole-file write calls.
403    #[must_use]
404    pub fn write_calls(&self) -> u64 {
405        self.counters.write_calls.load(Ordering::Relaxed)
406    }
407
408    /// Returns exact bytes supplied to successful whole-file writes.
409    #[must_use]
410    pub fn bytes_written(&self) -> u64 {
411        self.counters.bytes_written.load(Ordering::Relaxed)
412    }
413
414    /// Returns observed successful handle append calls.
415    #[must_use]
416    pub fn append_calls(&self) -> u64 {
417        self.counters.append_calls.load(Ordering::Relaxed)
418    }
419
420    /// Returns exact bytes supplied to successful handle appends.
421    #[must_use]
422    pub fn bytes_appended(&self) -> u64 {
423        self.counters.bytes_appended.load(Ordering::Relaxed)
424    }
425
426    /// Returns observed successful rename calls.
427    #[must_use]
428    pub fn rename_calls(&self) -> u64 {
429        self.counters.rename_calls.load(Ordering::Relaxed)
430    }
431
432    /// Returns observed successful delete calls.
433    #[must_use]
434    pub fn delete_calls(&self) -> u64 {
435        self.counters.delete_calls.load(Ordering::Relaxed)
436    }
437
438    /// Returns observed streaming directory-enumeration calls.
439    #[must_use]
440    pub fn enumerate_calls(&self) -> u64 {
441        self.counters.enumerate_calls.load(Ordering::Relaxed)
442    }
443
444    /// Returns path-based barrier synchronization calls.
445    #[must_use]
446    pub fn barrier_sync_calls(&self) -> u64 {
447        self.counters.barrier_sync_calls.load(Ordering::Relaxed)
448    }
449
450    /// Returns path-based full synchronization calls.
451    #[must_use]
452    pub fn full_sync_calls(&self) -> u64 {
453        self.counters.full_sync_calls.load(Ordering::Relaxed)
454    }
455
456    /// Returns open-handle barrier synchronization calls.
457    #[must_use]
458    pub fn handle_barrier_sync_calls(&self) -> u64 {
459        self.counters
460            .handle_barrier_sync_calls
461            .load(Ordering::Relaxed)
462    }
463
464    /// Returns open-handle full synchronization calls.
465    #[must_use]
466    pub fn handle_full_sync_calls(&self) -> u64 {
467        self.counters.handle_full_sync_calls.load(Ordering::Relaxed)
468    }
469
470    /// Resets all counters without changing the wrapped filesystem.
471    pub fn reset(&self) {
472        self.counters.open_calls.store(0, Ordering::Relaxed);
473        self.counters.open_for_map_calls.store(0, Ordering::Relaxed);
474        self.counters.read_calls.store(0, Ordering::Relaxed);
475        self.counters
476            .filtered_read_calls
477            .store(0, Ordering::Relaxed);
478        self.counters.read_bytes.store(0, Ordering::Relaxed);
479        self.counters.segment_bytes_read.store(0, Ordering::Relaxed);
480        self.counters.write_calls.store(0, Ordering::Relaxed);
481        self.counters.bytes_written.store(0, Ordering::Relaxed);
482        self.counters.append_calls.store(0, Ordering::Relaxed);
483        self.counters.bytes_appended.store(0, Ordering::Relaxed);
484        self.counters.rename_calls.store(0, Ordering::Relaxed);
485        self.counters.delete_calls.store(0, Ordering::Relaxed);
486        self.counters.enumerate_calls.store(0, Ordering::Relaxed);
487        self.counters.barrier_sync_calls.store(0, Ordering::Relaxed);
488        self.counters.full_sync_calls.store(0, Ordering::Relaxed);
489        self.counters
490            .handle_barrier_sync_calls
491            .store(0, Ordering::Relaxed);
492        self.counters
493            .handle_full_sync_calls
494            .store(0, Ordering::Relaxed);
495    }
496}
497
498impl<V: Vfs> Vfs for CountingVfs<V> {
499    fn truncate(&self, path: &Path, length: u64) -> std::io::Result<()> {
500        self.inner.truncate(path, length)
501    }
502    fn create_new(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
503        self.inner.create_new(path, bytes)?;
504        self.counters.write_calls.fetch_add(1, Ordering::Relaxed);
505        self.counters
506            .bytes_written
507            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
508        Ok(())
509    }
510
511    fn segment_data_read_counter(&self) -> Option<Arc<AtomicU64>> {
512        Some(Arc::clone(&self.counters.segment_bytes_read))
513    }
514
515    fn create_directory(&self, path: &Path) -> std::io::Result<()> {
516        self.inner.create_directory(path)
517    }
518    fn remove_directory(&self, path: &Path) -> std::io::Result<()> {
519        self.inner.remove_directory(path)
520    }
521    fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool> {
522        self.inner.ensure_directory(path, create)
523    }
524
525    fn open(&self, path: &Path) -> std::io::Result<u64> {
526        self.counters.open_calls.fetch_add(1, Ordering::Relaxed);
527        self.inner.open(path)
528    }
529
530    fn open_for_map(&self, path: &Path) -> std::io::Result<File> {
531        self.counters
532            .open_for_map_calls
533            .fetch_add(1, Ordering::Relaxed);
534        self.inner.open_for_map(path)
535    }
536
537    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
538        let bytes = self.inner.read(path)?;
539        self.counters.read_calls.fetch_add(1, Ordering::Relaxed);
540        if self.read_path_filter.as_deref() == Some(path) {
541            self.counters
542                .filtered_read_calls
543                .fetch_add(1, Ordering::Relaxed);
544        }
545        self.counters
546            .read_bytes
547            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
548        if is_segment_path(path) {
549            self.counters
550                .segment_bytes_read
551                .fetch_add(bytes.len() as u64, Ordering::Relaxed);
552        }
553        Ok(bytes)
554    }
555
556    fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>> {
557        let bytes = self.inner.read_range(path, offset, length)?;
558        self.counters.read_calls.fetch_add(1, Ordering::Relaxed);
559        if self.read_path_filter.as_deref() == Some(path) {
560            self.counters
561                .filtered_read_calls
562                .fetch_add(1, Ordering::Relaxed);
563        }
564        self.counters
565            .read_bytes
566            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
567        if is_segment_path(path) {
568            self.counters
569                .segment_bytes_read
570                .fetch_add(bytes.len() as u64, Ordering::Relaxed);
571        }
572        Ok(bytes)
573    }
574
575    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
576        self.inner.write(path, bytes)?;
577        self.counters.write_calls.fetch_add(1, Ordering::Relaxed);
578        self.counters
579            .bytes_written
580            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
581        Ok(())
582    }
583
584    fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>> {
585        Ok(Box::new(CountingVfsFile {
586            inner: self.inner.open_append(path)?,
587            counters: Arc::clone(&self.counters),
588        }))
589    }
590
591    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
592        self.inner.rename(from, to)?;
593        self.counters.rename_calls.fetch_add(1, Ordering::Relaxed);
594        Ok(())
595    }
596
597    fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()> {
598        self.inner.sync(path, kind)?;
599        match kind {
600            SyncKind::Barrier => &self.counters.barrier_sync_calls,
601            SyncKind::Full => &self.counters.full_sync_calls,
602        }
603        .fetch_add(1, Ordering::Relaxed);
604        Ok(())
605    }
606
607    fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>> {
608        self.inner.list(directory)
609    }
610
611    fn for_each_direct_child(
612        &self,
613        directory: &Path,
614        visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
615    ) -> std::io::Result<()> {
616        self.inner.for_each_direct_child(directory, visitor)?;
617        self.counters
618            .enumerate_calls
619            .fetch_add(1, Ordering::Relaxed);
620        Ok(())
621    }
622
623    fn delete(&self, path: &Path) -> std::io::Result<()> {
624        self.inner.delete(path)?;
625        self.counters.delete_calls.fetch_add(1, Ordering::Relaxed);
626        Ok(())
627    }
628}
629
630fn is_segment_path(path: &Path) -> bool {
631    path.file_name()
632        .and_then(|name| name.to_str())
633        .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".zseg"))
634}
635
636/// Synchronizes whatever `path` names, reopening it with the access the
637/// platform requires.
638///
639/// On Unix a read-only descriptor is enough for `fsync`, including on a
640/// directory. Windows refuses both: `FlushFileBuffers` needs write access, and
641/// a directory handle additionally needs `FILE_FLAG_BACKUP_SEMANTICS`, so the
642/// Windows arm does its own open rather than reusing `File::open`.
643#[cfg(not(windows))]
644fn sync_path(path: &Path, kind: SyncKind) -> std::io::Result<()> {
645    sync_file(&File::open(path)?, kind)
646}
647
648#[cfg(windows)]
649fn sync_path(path: &Path, _kind: SyncKind) -> std::io::Result<()> {
650    // Windows exposes no ordering-only barrier, so `Barrier` is served by the
651    // same full flush as `Full`: stronger and more expensive than a barrier,
652    // never cheaper. See `tasks/evidence/windows/w02-durable-publication-protocol.md`.
653    crate::sys::windows::sync_path(path)
654}
655
656#[cfg(any(target_os = "macos", target_os = "ios"))]
657fn sync_file(file: &File, kind: SyncKind) -> std::io::Result<()> {
658    let result = match kind {
659        SyncKind::Barrier => crate::sys::darwin::barrier_fsync(file.as_raw_fd()),
660        SyncKind::Full => crate::sys::darwin::full_fsync(file.as_raw_fd()),
661    };
662    result.map_err(std::io::Error::other)
663}
664
665#[cfg(all(unix, not(any(target_os = "macos", target_os = "ios"))))]
666fn sync_file(file: &File, kind: SyncKind) -> std::io::Result<()> {
667    let result = unsafe {
668        // SAFETY: both calls accept an owned live descriptor and report failures through errno.
669        match kind {
670            SyncKind::Barrier => libc::fdatasync(file.as_raw_fd()),
671            SyncKind::Full => libc::fsync(file.as_raw_fd()),
672        }
673    };
674    if result == -1 {
675        Err(std::io::Error::last_os_error())
676    } else {
677        Ok(())
678    }
679}
680
681/// Flushes an already-open handle without reopening its path.
682///
683/// Both `SyncKind`s map to `FlushFileBuffers`: Windows has no barrier-only
684/// primitive, so `Ordered` durability is served by the full flush. That is
685/// documented as stronger and more expensive, not as a cheap barrier.
686#[cfg(windows)]
687fn sync_file(file: &File, _kind: SyncKind) -> std::io::Result<()> {
688    crate::sys::windows::flush_file(file)
689}
690
691#[cfg(not(any(unix, windows)))]
692fn sync_file(_: &File, kind: SyncKind) -> std::io::Result<()> {
693    Err(std::io::Error::new(
694        std::io::ErrorKind::Unsupported,
695        format!("{kind:?} synchronization requires a supported file-descriptor platform"),
696    ))
697}
698
699#[cfg(test)]
700mod tests {
701    use std::path::Path;
702
703    use super::crash::MemoryVfs;
704    use super::{CountingVfs, SyncKind, Vfs};
705
706    #[test]
707    fn crash_support_api_gate_is_identical_in_debug_and_release() {
708        let source = include_str!("mod.rs");
709        let gate = source
710            .lines()
711            .zip(source.lines().skip(1))
712            .find_map(|(line, next)| (next == "pub mod crash;").then_some(line));
713        assert_eq!(gate, Some("#[cfg(any(test, feature = \"test-seams\"))]"));
714    }
715
716    #[test]
717    #[allow(clippy::expect_used)]
718    fn counting_vfs_counts_every_mutation_and_sync_kind_exactly() {
719        let inner = MemoryVfs::new();
720        inner
721            .insert("/source", b"abc".to_vec())
722            .expect("seed source");
723        let counting = CountingVfs::new(inner);
724
725        assert_eq!(counting.open(Path::new("/source")).expect("open"), 3);
726        assert_eq!(counting.read(Path::new("/source")).expect("read"), b"abc");
727        assert_eq!(
728            counting
729                .read_range(Path::new("/source"), 1, 2)
730                .expect("range"),
731            b"bc"
732        );
733        counting.write(Path::new("/write"), b"1234").expect("write");
734        let mut handle = counting
735            .open_append(Path::new("/append"))
736            .expect("open append");
737        handle.append(b"12").expect("first append");
738        handle.append(b"345").expect("second append");
739        handle.sync(SyncKind::Barrier).expect("handle barrier");
740        handle.sync(SyncKind::Full).expect("handle full");
741        counting
742            .sync(Path::new("/write"), SyncKind::Barrier)
743            .expect("path barrier");
744        counting
745            .sync(Path::new("/write"), SyncKind::Full)
746            .expect("path full");
747        counting
748            .rename(Path::new("/write"), Path::new("/renamed"))
749            .expect("rename");
750        counting.delete(Path::new("/renamed")).expect("delete");
751
752        assert_eq!(
753            [
754                counting.open_calls(),
755                counting.read_calls(),
756                counting.read_bytes(),
757                counting.write_calls(),
758                counting.bytes_written(),
759                counting.append_calls(),
760                counting.bytes_appended(),
761                counting.rename_calls(),
762                counting.delete_calls(),
763                counting.barrier_sync_calls(),
764                counting.full_sync_calls(),
765                counting.handle_barrier_sync_calls(),
766                counting.handle_full_sync_calls(),
767            ],
768            [1, 2, 5, 1, 4, 2, 5, 1, 1, 1, 1, 1, 1],
769            "CountingVfs must expose every exact VFS and open-handle cost"
770        );
771        counting.reset();
772        assert_eq!(
773            [
774                counting.open_calls(),
775                counting.read_calls(),
776                counting.read_bytes(),
777                counting.write_calls(),
778                counting.bytes_written(),
779                counting.append_calls(),
780                counting.bytes_appended(),
781                counting.rename_calls(),
782                counting.delete_calls(),
783                counting.barrier_sync_calls(),
784                counting.full_sync_calls(),
785                counting.handle_barrier_sync_calls(),
786                counting.handle_full_sync_calls(),
787            ],
788            [0; 13],
789            "reset must zero decorator and shared-handle counters"
790        );
791    }
792}
793
794#[cfg(feature = "test-seams")]
795pub mod file_test_support;