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        #[cfg(feature = "test-seams")]
137        file_test_support::event(file_test_support::FileEvent::BeforeAppend)?;
138        let mut remaining = buffers;
139        while !remaining.is_empty() {
140            let written = self.0.write_vectored(remaining)?;
141            if written == 0 {
142                return Err(std::io::Error::new(
143                    std::io::ErrorKind::WriteZero,
144                    "failed to append vectored WAL group",
145                ));
146            }
147            IoSlice::advance_slices(&mut remaining, written);
148        }
149        Ok(())
150    }
151
152    fn sync(&self, kind: SyncKind) -> std::io::Result<()> {
153        #[cfg(feature = "test-seams")]
154        file_test_support::event(file_test_support::FileEvent::BeforeSync)?;
155        sync_file(&self.0, kind)?;
156        #[cfg(feature = "test-seams")]
157        file_test_support::event(file_test_support::FileEvent::AfterSync)?;
158        Ok(())
159    }
160}
161
162impl Vfs for StdVfs {
163    fn truncate(&self, path: &Path, length: u64) -> std::io::Result<()> {
164        OpenOptions::new().write(true).open(path)?.set_len(length)
165    }
166
167    fn create_new(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
168        let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
169        file.write_all(bytes)
170    }
171
172    fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool> {
173        if create {
174            std::fs::create_dir_all(path)?;
175        }
176        Ok(std::fs::metadata(path)?.is_dir())
177    }
178
179    fn create_directory(&self, path: &Path) -> std::io::Result<()> {
180        std::fs::create_dir(path)
181    }
182
183    fn remove_directory(&self, path: &Path) -> std::io::Result<()> {
184        std::fs::remove_dir(path)
185    }
186
187    fn open(&self, path: &Path) -> std::io::Result<u64> {
188        Ok(File::open(path)?.metadata()?.len())
189    }
190
191    fn open_for_map(&self, path: &Path) -> std::io::Result<File> {
192        File::open(path)
193    }
194
195    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
196        let mut file = File::open(path)?;
197        let length = usize::try_from(file.metadata()?.len()).map_err(|_| {
198            std::io::Error::new(std::io::ErrorKind::InvalidData, "file length exceeds usize")
199        })?;
200        let mut bytes = Vec::with_capacity(length);
201        file.read_to_end(&mut bytes)?;
202        Ok(bytes)
203    }
204
205    fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>> {
206        let mut file = File::open(path)?;
207        file.seek(SeekFrom::Start(offset))?;
208        let mut bytes = vec![0_u8; length];
209        let mut filled = 0_usize;
210        while filled < length {
211            let remaining = bytes.get_mut(filled..).ok_or_else(|| {
212                std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid read range")
213            })?;
214            let read = file.read(remaining)?;
215            if read == 0 {
216                break;
217            }
218            filled = filled.saturating_add(read);
219        }
220        bytes.truncate(filled);
221        Ok(bytes)
222    }
223
224    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
225        let mut file = OpenOptions::new()
226            .create(true)
227            .truncate(true)
228            .write(true)
229            .open(path)?;
230        file.write_all(bytes)
231    }
232
233    fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>> {
234        let file = OpenOptions::new().create(true).append(true).open(path)?;
235        Ok(Box::new(StdVfsFile(file)))
236    }
237
238    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
239        std::fs::rename(from, to)
240    }
241
242    fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()> {
243        sync_path(path, kind)
244    }
245
246    fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>> {
247        std::fs::read_dir(directory)?
248            .map(|entry| entry.map(|value| value.path()))
249            .collect()
250    }
251
252    fn for_each_direct_child(
253        &self,
254        directory: &Path,
255        visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
256    ) -> std::io::Result<()> {
257        for entry in std::fs::read_dir(directory)? {
258            let entry = entry?;
259            visitor(&entry.path())?;
260        }
261        Ok(())
262    }
263
264    fn delete(&self, path: &Path) -> std::io::Result<()> {
265        std::fs::remove_file(path)
266    }
267}
268
269/// Byte/call-counting decorator used by deterministic open-cost gates.
270pub struct CountingVfs<V> {
271    inner: V,
272    counters: Arc<CountingVfsCounters>,
273    read_path_filter: Option<PathBuf>,
274}
275
276#[derive(Default)]
277struct CountingVfsCounters {
278    open_calls: AtomicU64,
279    open_for_map_calls: AtomicU64,
280    read_calls: AtomicU64,
281    filtered_read_calls: AtomicU64,
282    read_bytes: AtomicU64,
283    segment_bytes_read: Arc<AtomicU64>,
284    write_calls: AtomicU64,
285    bytes_written: AtomicU64,
286    append_calls: AtomicU64,
287    bytes_appended: AtomicU64,
288    rename_calls: AtomicU64,
289    delete_calls: AtomicU64,
290    enumerate_calls: AtomicU64,
291    barrier_sync_calls: AtomicU64,
292    full_sync_calls: AtomicU64,
293    handle_barrier_sync_calls: AtomicU64,
294    handle_full_sync_calls: AtomicU64,
295}
296
297struct CountingVfsFile {
298    inner: Box<dyn VfsFile>,
299    counters: Arc<CountingVfsCounters>,
300}
301
302impl VfsFile for CountingVfsFile {
303    fn append(&mut self, bytes: &[u8]) -> std::io::Result<()> {
304        self.inner.append(bytes)?;
305        self.counters.append_calls.fetch_add(1, Ordering::Relaxed);
306        self.counters
307            .bytes_appended
308            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
309        Ok(())
310    }
311
312    fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
313        let bytes = buffers
314            .iter()
315            .fold(0_usize, |total, buffer| total.saturating_add(buffer.len()));
316        self.inner.append_vectored(buffers)?;
317        self.counters.append_calls.fetch_add(1, Ordering::Relaxed);
318        self.counters
319            .bytes_appended
320            .fetch_add(bytes as u64, Ordering::Relaxed);
321        Ok(())
322    }
323
324    fn sync(&self, kind: SyncKind) -> std::io::Result<()> {
325        self.inner.sync(kind)?;
326        match kind {
327            SyncKind::Barrier => &self.counters.handle_barrier_sync_calls,
328            SyncKind::Full => &self.counters.handle_full_sync_calls,
329        }
330        .fetch_add(1, Ordering::Relaxed);
331        Ok(())
332    }
333}
334
335impl<V> CountingVfs<V> {
336    /// Wraps a filesystem with zeroed counters.
337    #[must_use]
338    pub fn new(inner: V) -> Self {
339        Self {
340            inner,
341            counters: Arc::new(CountingVfsCounters::default()),
342            read_path_filter: None,
343        }
344    }
345
346    /// Counts successful reads of one exact path separately from global reads.
347    #[must_use]
348    pub fn with_read_path_filter(mut self, path: impl Into<PathBuf>) -> Self {
349        self.read_path_filter = Some(path.into());
350        self
351    }
352
353    /// Returns the wrapped filesystem.
354    #[must_use]
355    pub const fn inner(&self) -> &V {
356        &self.inner
357    }
358
359    /// Runs a test probe with mmap-backed segment payload accounting enabled.
360    #[doc(hidden)]
361    pub fn audit_segment_data_reads<T>(&self, operation: impl FnOnce() -> T) -> T {
362        let _audit = crate::segment::reader::install_data_read_audit(Some(Arc::clone(
363            &self.counters.segment_bytes_read,
364        )));
365        operation()
366    }
367
368    /// Returns observed open calls.
369    #[must_use]
370    pub fn open_calls(&self) -> u64 {
371        self.counters.open_calls.load(Ordering::Relaxed)
372    }
373
374    /// Returns observed mapping-file open calls.
375    #[must_use]
376    pub fn open_for_map_calls(&self) -> u64 {
377        self.counters.open_for_map_calls.load(Ordering::Relaxed)
378    }
379
380    /// Returns observed read calls.
381    #[must_use]
382    pub fn read_calls(&self) -> u64 {
383        self.counters.read_calls.load(Ordering::Relaxed)
384    }
385
386    /// Returns successful reads matching the configured exact path.
387    #[must_use]
388    pub fn filtered_read_calls(&self) -> u64 {
389        self.counters.filtered_read_calls.load(Ordering::Relaxed)
390    }
391
392    /// Returns exact bytes returned by reads.
393    #[must_use]
394    pub fn read_bytes(&self) -> u64 {
395        self.counters.read_bytes.load(Ordering::Relaxed)
396    }
397
398    /// Returns exact bytes read from immutable segment paths.
399    #[must_use]
400    pub fn segment_bytes_read(&self) -> u64 {
401        self.counters.segment_bytes_read.load(Ordering::Relaxed)
402    }
403
404    /// Returns observed whole-file write calls.
405    #[must_use]
406    pub fn write_calls(&self) -> u64 {
407        self.counters.write_calls.load(Ordering::Relaxed)
408    }
409
410    /// Returns exact bytes supplied to successful whole-file writes.
411    #[must_use]
412    pub fn bytes_written(&self) -> u64 {
413        self.counters.bytes_written.load(Ordering::Relaxed)
414    }
415
416    /// Returns observed successful handle append calls.
417    #[must_use]
418    pub fn append_calls(&self) -> u64 {
419        self.counters.append_calls.load(Ordering::Relaxed)
420    }
421
422    /// Returns exact bytes supplied to successful handle appends.
423    #[must_use]
424    pub fn bytes_appended(&self) -> u64 {
425        self.counters.bytes_appended.load(Ordering::Relaxed)
426    }
427
428    /// Returns observed successful rename calls.
429    #[must_use]
430    pub fn rename_calls(&self) -> u64 {
431        self.counters.rename_calls.load(Ordering::Relaxed)
432    }
433
434    /// Returns observed successful delete calls.
435    #[must_use]
436    pub fn delete_calls(&self) -> u64 {
437        self.counters.delete_calls.load(Ordering::Relaxed)
438    }
439
440    /// Returns observed streaming directory-enumeration calls.
441    #[must_use]
442    pub fn enumerate_calls(&self) -> u64 {
443        self.counters.enumerate_calls.load(Ordering::Relaxed)
444    }
445
446    /// Returns path-based barrier synchronization calls.
447    #[must_use]
448    pub fn barrier_sync_calls(&self) -> u64 {
449        self.counters.barrier_sync_calls.load(Ordering::Relaxed)
450    }
451
452    /// Returns path-based full synchronization calls.
453    #[must_use]
454    pub fn full_sync_calls(&self) -> u64 {
455        self.counters.full_sync_calls.load(Ordering::Relaxed)
456    }
457
458    /// Returns open-handle barrier synchronization calls.
459    #[must_use]
460    pub fn handle_barrier_sync_calls(&self) -> u64 {
461        self.counters
462            .handle_barrier_sync_calls
463            .load(Ordering::Relaxed)
464    }
465
466    /// Returns open-handle full synchronization calls.
467    #[must_use]
468    pub fn handle_full_sync_calls(&self) -> u64 {
469        self.counters.handle_full_sync_calls.load(Ordering::Relaxed)
470    }
471
472    /// Resets all counters without changing the wrapped filesystem.
473    pub fn reset(&self) {
474        self.counters.open_calls.store(0, Ordering::Relaxed);
475        self.counters.open_for_map_calls.store(0, Ordering::Relaxed);
476        self.counters.read_calls.store(0, Ordering::Relaxed);
477        self.counters
478            .filtered_read_calls
479            .store(0, Ordering::Relaxed);
480        self.counters.read_bytes.store(0, Ordering::Relaxed);
481        self.counters.segment_bytes_read.store(0, Ordering::Relaxed);
482        self.counters.write_calls.store(0, Ordering::Relaxed);
483        self.counters.bytes_written.store(0, Ordering::Relaxed);
484        self.counters.append_calls.store(0, Ordering::Relaxed);
485        self.counters.bytes_appended.store(0, Ordering::Relaxed);
486        self.counters.rename_calls.store(0, Ordering::Relaxed);
487        self.counters.delete_calls.store(0, Ordering::Relaxed);
488        self.counters.enumerate_calls.store(0, Ordering::Relaxed);
489        self.counters.barrier_sync_calls.store(0, Ordering::Relaxed);
490        self.counters.full_sync_calls.store(0, Ordering::Relaxed);
491        self.counters
492            .handle_barrier_sync_calls
493            .store(0, Ordering::Relaxed);
494        self.counters
495            .handle_full_sync_calls
496            .store(0, Ordering::Relaxed);
497    }
498}
499
500impl<V: Vfs> Vfs for CountingVfs<V> {
501    fn truncate(&self, path: &Path, length: u64) -> std::io::Result<()> {
502        self.inner.truncate(path, length)
503    }
504    fn create_new(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
505        self.inner.create_new(path, bytes)?;
506        self.counters.write_calls.fetch_add(1, Ordering::Relaxed);
507        self.counters
508            .bytes_written
509            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
510        Ok(())
511    }
512
513    fn segment_data_read_counter(&self) -> Option<Arc<AtomicU64>> {
514        Some(Arc::clone(&self.counters.segment_bytes_read))
515    }
516
517    fn create_directory(&self, path: &Path) -> std::io::Result<()> {
518        self.inner.create_directory(path)
519    }
520    fn remove_directory(&self, path: &Path) -> std::io::Result<()> {
521        self.inner.remove_directory(path)
522    }
523    fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool> {
524        self.inner.ensure_directory(path, create)
525    }
526
527    fn open(&self, path: &Path) -> std::io::Result<u64> {
528        self.counters.open_calls.fetch_add(1, Ordering::Relaxed);
529        self.inner.open(path)
530    }
531
532    fn open_for_map(&self, path: &Path) -> std::io::Result<File> {
533        self.counters
534            .open_for_map_calls
535            .fetch_add(1, Ordering::Relaxed);
536        self.inner.open_for_map(path)
537    }
538
539    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
540        let bytes = self.inner.read(path)?;
541        self.counters.read_calls.fetch_add(1, Ordering::Relaxed);
542        if self.read_path_filter.as_deref() == Some(path) {
543            self.counters
544                .filtered_read_calls
545                .fetch_add(1, Ordering::Relaxed);
546        }
547        self.counters
548            .read_bytes
549            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
550        if is_segment_path(path) {
551            self.counters
552                .segment_bytes_read
553                .fetch_add(bytes.len() as u64, Ordering::Relaxed);
554        }
555        Ok(bytes)
556    }
557
558    fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>> {
559        let bytes = self.inner.read_range(path, offset, length)?;
560        self.counters.read_calls.fetch_add(1, Ordering::Relaxed);
561        if self.read_path_filter.as_deref() == Some(path) {
562            self.counters
563                .filtered_read_calls
564                .fetch_add(1, Ordering::Relaxed);
565        }
566        self.counters
567            .read_bytes
568            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
569        if is_segment_path(path) {
570            self.counters
571                .segment_bytes_read
572                .fetch_add(bytes.len() as u64, Ordering::Relaxed);
573        }
574        Ok(bytes)
575    }
576
577    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
578        self.inner.write(path, bytes)?;
579        self.counters.write_calls.fetch_add(1, Ordering::Relaxed);
580        self.counters
581            .bytes_written
582            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
583        Ok(())
584    }
585
586    fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>> {
587        Ok(Box::new(CountingVfsFile {
588            inner: self.inner.open_append(path)?,
589            counters: Arc::clone(&self.counters),
590        }))
591    }
592
593    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
594        self.inner.rename(from, to)?;
595        self.counters.rename_calls.fetch_add(1, Ordering::Relaxed);
596        Ok(())
597    }
598
599    fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()> {
600        self.inner.sync(path, kind)?;
601        match kind {
602            SyncKind::Barrier => &self.counters.barrier_sync_calls,
603            SyncKind::Full => &self.counters.full_sync_calls,
604        }
605        .fetch_add(1, Ordering::Relaxed);
606        Ok(())
607    }
608
609    fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>> {
610        self.inner.list(directory)
611    }
612
613    fn for_each_direct_child(
614        &self,
615        directory: &Path,
616        visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
617    ) -> std::io::Result<()> {
618        self.inner.for_each_direct_child(directory, visitor)?;
619        self.counters
620            .enumerate_calls
621            .fetch_add(1, Ordering::Relaxed);
622        Ok(())
623    }
624
625    fn delete(&self, path: &Path) -> std::io::Result<()> {
626        self.inner.delete(path)?;
627        self.counters.delete_calls.fetch_add(1, Ordering::Relaxed);
628        Ok(())
629    }
630}
631
632fn is_segment_path(path: &Path) -> bool {
633    path.file_name()
634        .and_then(|name| name.to_str())
635        .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".zseg"))
636}
637
638/// Synchronizes whatever `path` names, reopening it with the access the
639/// platform requires.
640///
641/// On Unix a read-only descriptor is enough for `fsync`, including on a
642/// directory. Windows refuses both: `FlushFileBuffers` needs write access, and
643/// a directory handle additionally needs `FILE_FLAG_BACKUP_SEMANTICS`, so the
644/// Windows arm does its own open rather than reusing `File::open`.
645#[cfg(not(windows))]
646fn sync_path(path: &Path, kind: SyncKind) -> std::io::Result<()> {
647    sync_file(&File::open(path)?, kind)
648}
649
650#[cfg(windows)]
651fn sync_path(path: &Path, _kind: SyncKind) -> std::io::Result<()> {
652    // Windows exposes no ordering-only barrier, so `Barrier` is served by the
653    // same full flush as `Full`: stronger and more expensive than a barrier,
654    // never cheaper. See `tasks/evidence/windows/w02-durable-publication-protocol.md`.
655    crate::sys::windows::sync_path(path)
656}
657
658#[cfg(any(target_os = "macos", target_os = "ios"))]
659fn sync_file(file: &File, kind: SyncKind) -> std::io::Result<()> {
660    let result = match kind {
661        SyncKind::Barrier => crate::sys::darwin::barrier_fsync(file.as_raw_fd()),
662        SyncKind::Full => crate::sys::darwin::full_fsync(file.as_raw_fd()),
663    };
664    result.map_err(std::io::Error::other)
665}
666
667#[cfg(all(unix, not(any(target_os = "macos", target_os = "ios"))))]
668fn sync_file(file: &File, kind: SyncKind) -> std::io::Result<()> {
669    let result = unsafe {
670        // SAFETY: both calls accept an owned live descriptor and report failures through errno.
671        match kind {
672            SyncKind::Barrier => libc::fdatasync(file.as_raw_fd()),
673            SyncKind::Full => libc::fsync(file.as_raw_fd()),
674        }
675    };
676    if result == -1 {
677        Err(std::io::Error::last_os_error())
678    } else {
679        Ok(())
680    }
681}
682
683/// Flushes an already-open handle without reopening its path.
684///
685/// Both `SyncKind`s map to `FlushFileBuffers`: Windows has no barrier-only
686/// primitive, so `Ordered` durability is served by the full flush. That is
687/// documented as stronger and more expensive, not as a cheap barrier.
688#[cfg(windows)]
689fn sync_file(file: &File, _kind: SyncKind) -> std::io::Result<()> {
690    crate::sys::windows::flush_file(file)
691}
692
693#[cfg(not(any(unix, windows)))]
694fn sync_file(_: &File, kind: SyncKind) -> std::io::Result<()> {
695    Err(std::io::Error::new(
696        std::io::ErrorKind::Unsupported,
697        format!("{kind:?} synchronization requires a supported file-descriptor platform"),
698    ))
699}
700
701#[cfg(test)]
702mod tests {
703    use std::path::Path;
704
705    use super::crash::MemoryVfs;
706    use super::{CountingVfs, SyncKind, Vfs};
707
708    #[test]
709    fn crash_support_api_gate_is_identical_in_debug_and_release() {
710        let source = include_str!("mod.rs");
711        let gate = source
712            .lines()
713            .zip(source.lines().skip(1))
714            .find_map(|(line, next)| (next == "pub mod crash;").then_some(line));
715        assert_eq!(gate, Some("#[cfg(any(test, feature = \"test-seams\"))]"));
716    }
717
718    #[test]
719    #[allow(clippy::expect_used)]
720    fn counting_vfs_counts_every_mutation_and_sync_kind_exactly() {
721        let inner = MemoryVfs::new();
722        inner
723            .insert("/source", b"abc".to_vec())
724            .expect("seed source");
725        let counting = CountingVfs::new(inner);
726
727        assert_eq!(counting.open(Path::new("/source")).expect("open"), 3);
728        assert_eq!(counting.read(Path::new("/source")).expect("read"), b"abc");
729        assert_eq!(
730            counting
731                .read_range(Path::new("/source"), 1, 2)
732                .expect("range"),
733            b"bc"
734        );
735        counting.write(Path::new("/write"), b"1234").expect("write");
736        let mut handle = counting
737            .open_append(Path::new("/append"))
738            .expect("open append");
739        handle.append(b"12").expect("first append");
740        handle.append(b"345").expect("second append");
741        handle.sync(SyncKind::Barrier).expect("handle barrier");
742        handle.sync(SyncKind::Full).expect("handle full");
743        counting
744            .sync(Path::new("/write"), SyncKind::Barrier)
745            .expect("path barrier");
746        counting
747            .sync(Path::new("/write"), SyncKind::Full)
748            .expect("path full");
749        counting
750            .rename(Path::new("/write"), Path::new("/renamed"))
751            .expect("rename");
752        counting.delete(Path::new("/renamed")).expect("delete");
753
754        assert_eq!(
755            [
756                counting.open_calls(),
757                counting.read_calls(),
758                counting.read_bytes(),
759                counting.write_calls(),
760                counting.bytes_written(),
761                counting.append_calls(),
762                counting.bytes_appended(),
763                counting.rename_calls(),
764                counting.delete_calls(),
765                counting.barrier_sync_calls(),
766                counting.full_sync_calls(),
767                counting.handle_barrier_sync_calls(),
768                counting.handle_full_sync_calls(),
769            ],
770            [1, 2, 5, 1, 4, 2, 5, 1, 1, 1, 1, 1, 1],
771            "CountingVfs must expose every exact VFS and open-handle cost"
772        );
773        counting.reset();
774        assert_eq!(
775            [
776                counting.open_calls(),
777                counting.read_calls(),
778                counting.read_bytes(),
779                counting.write_calls(),
780                counting.bytes_written(),
781                counting.append_calls(),
782                counting.bytes_appended(),
783                counting.rename_calls(),
784                counting.delete_calls(),
785                counting.barrier_sync_calls(),
786                counting.full_sync_calls(),
787                counting.handle_barrier_sync_calls(),
788                counting.handle_full_sync_calls(),
789            ],
790            [0; 13],
791            "reset must zero decorator and shared-handle counters"
792        );
793    }
794}
795
796#[cfg(feature = "test-seams")]
797pub mod file_test_support;