#[cfg(any(test, feature = "test-seams"))]
pub mod crash;
#[cfg(any(test, feature = "test-seams"))]
pub mod fault;
use std::fs::{File, OpenOptions};
use std::io::{IoSlice, Read, Seek, SeekFrom, Write};
#[cfg(unix)]
use std::os::fd::AsRawFd;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SyncKind {
Barrier,
Full,
}
pub trait VfsFile: Send {
fn append(&mut self, bytes: &[u8]) -> std::io::Result<()>;
fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
for buffer in buffers {
self.append(buffer)?;
}
Ok(())
}
fn sync(&self, kind: SyncKind) -> std::io::Result<()>;
}
pub trait Vfs: Send + Sync {
#[doc(hidden)]
fn segment_data_read_counter(&self) -> Option<Arc<AtomicU64>> {
None
}
fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool>;
fn create_directory(&self, _path: &Path) -> std::io::Result<()> {
Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"VFS does not support exclusive directory creation",
))
}
fn remove_directory(&self, _path: &Path) -> std::io::Result<()> {
Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"VFS does not support directory removal",
))
}
fn open(&self, path: &Path) -> std::io::Result<u64>;
fn open_for_map(&self, path: &Path) -> std::io::Result<File>;
fn read(&self, path: &Path) -> std::io::Result<Vec<u8>>;
fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>>;
fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()>;
fn truncate(&self, _path: &Path, _length: u64) -> std::io::Result<()> {
Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"VFS does not support suffix truncation",
))
}
fn create_new(&self, _path: &Path, _bytes: &[u8]) -> std::io::Result<()> {
Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"VFS does not support exclusive file creation",
))
}
fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>>;
fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()>;
fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()>;
fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>>;
fn for_each_direct_child(
&self,
directory: &Path,
visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
) -> std::io::Result<()>;
fn delete(&self, path: &Path) -> std::io::Result<()>;
}
#[derive(Clone, Copy, Debug, Default)]
pub struct StdVfs;
struct StdVfsFile(File);
impl VfsFile for StdVfsFile {
fn append(&mut self, bytes: &[u8]) -> std::io::Result<()> {
#[cfg(feature = "test-seams")]
file_test_support::event(file_test_support::FileEvent::BeforeAppend)?;
self.0.write_all(bytes)
}
fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
#[cfg(feature = "test-seams")]
file_test_support::event(file_test_support::FileEvent::BeforeAppend)?;
let mut remaining = buffers;
while !remaining.is_empty() {
let written = self.0.write_vectored(remaining)?;
if written == 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::WriteZero,
"failed to append vectored WAL group",
));
}
IoSlice::advance_slices(&mut remaining, written);
}
Ok(())
}
fn sync(&self, kind: SyncKind) -> std::io::Result<()> {
#[cfg(feature = "test-seams")]
file_test_support::event(file_test_support::FileEvent::BeforeSync)?;
sync_file(&self.0, kind)?;
#[cfg(feature = "test-seams")]
file_test_support::event(file_test_support::FileEvent::AfterSync)?;
Ok(())
}
}
impl Vfs for StdVfs {
fn truncate(&self, path: &Path, length: u64) -> std::io::Result<()> {
OpenOptions::new().write(true).open(path)?.set_len(length)
}
fn create_new(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
file.write_all(bytes)
}
fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool> {
if create {
std::fs::create_dir_all(path)?;
}
Ok(std::fs::metadata(path)?.is_dir())
}
fn create_directory(&self, path: &Path) -> std::io::Result<()> {
std::fs::create_dir(path)
}
fn remove_directory(&self, path: &Path) -> std::io::Result<()> {
std::fs::remove_dir(path)
}
fn open(&self, path: &Path) -> std::io::Result<u64> {
Ok(File::open(path)?.metadata()?.len())
}
fn open_for_map(&self, path: &Path) -> std::io::Result<File> {
File::open(path)
}
fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
let mut file = File::open(path)?;
let length = usize::try_from(file.metadata()?.len()).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::InvalidData, "file length exceeds usize")
})?;
let mut bytes = Vec::with_capacity(length);
file.read_to_end(&mut bytes)?;
Ok(bytes)
}
fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>> {
let mut file = File::open(path)?;
file.seek(SeekFrom::Start(offset))?;
let mut bytes = vec![0_u8; length];
let mut filled = 0_usize;
while filled < length {
let remaining = bytes.get_mut(filled..).ok_or_else(|| {
std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid read range")
})?;
let read = file.read(remaining)?;
if read == 0 {
break;
}
filled = filled.saturating_add(read);
}
bytes.truncate(filled);
Ok(bytes)
}
fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
let mut file = OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(path)?;
file.write_all(bytes)
}
fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>> {
let file = OpenOptions::new().create(true).append(true).open(path)?;
Ok(Box::new(StdVfsFile(file)))
}
fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
std::fs::rename(from, to)
}
fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()> {
sync_path(path, kind)
}
fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>> {
std::fs::read_dir(directory)?
.map(|entry| entry.map(|value| value.path()))
.collect()
}
fn for_each_direct_child(
&self,
directory: &Path,
visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
) -> std::io::Result<()> {
for entry in std::fs::read_dir(directory)? {
let entry = entry?;
visitor(&entry.path())?;
}
Ok(())
}
fn delete(&self, path: &Path) -> std::io::Result<()> {
std::fs::remove_file(path)
}
}
pub struct CountingVfs<V> {
inner: V,
counters: Arc<CountingVfsCounters>,
read_path_filter: Option<PathBuf>,
}
#[derive(Default)]
struct CountingVfsCounters {
open_calls: AtomicU64,
open_for_map_calls: AtomicU64,
read_calls: AtomicU64,
filtered_read_calls: AtomicU64,
read_bytes: AtomicU64,
segment_bytes_read: Arc<AtomicU64>,
write_calls: AtomicU64,
bytes_written: AtomicU64,
append_calls: AtomicU64,
bytes_appended: AtomicU64,
rename_calls: AtomicU64,
delete_calls: AtomicU64,
enumerate_calls: AtomicU64,
barrier_sync_calls: AtomicU64,
full_sync_calls: AtomicU64,
handle_barrier_sync_calls: AtomicU64,
handle_full_sync_calls: AtomicU64,
}
struct CountingVfsFile {
inner: Box<dyn VfsFile>,
counters: Arc<CountingVfsCounters>,
}
impl VfsFile for CountingVfsFile {
fn append(&mut self, bytes: &[u8]) -> std::io::Result<()> {
self.inner.append(bytes)?;
self.counters.append_calls.fetch_add(1, Ordering::Relaxed);
self.counters
.bytes_appended
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
Ok(())
}
fn append_vectored(&mut self, buffers: &mut [IoSlice<'_>]) -> std::io::Result<()> {
let bytes = buffers
.iter()
.fold(0_usize, |total, buffer| total.saturating_add(buffer.len()));
self.inner.append_vectored(buffers)?;
self.counters.append_calls.fetch_add(1, Ordering::Relaxed);
self.counters
.bytes_appended
.fetch_add(bytes as u64, Ordering::Relaxed);
Ok(())
}
fn sync(&self, kind: SyncKind) -> std::io::Result<()> {
self.inner.sync(kind)?;
match kind {
SyncKind::Barrier => &self.counters.handle_barrier_sync_calls,
SyncKind::Full => &self.counters.handle_full_sync_calls,
}
.fetch_add(1, Ordering::Relaxed);
Ok(())
}
}
impl<V> CountingVfs<V> {
#[must_use]
pub fn new(inner: V) -> Self {
Self {
inner,
counters: Arc::new(CountingVfsCounters::default()),
read_path_filter: None,
}
}
#[must_use]
pub fn with_read_path_filter(mut self, path: impl Into<PathBuf>) -> Self {
self.read_path_filter = Some(path.into());
self
}
#[must_use]
pub const fn inner(&self) -> &V {
&self.inner
}
#[doc(hidden)]
pub fn audit_segment_data_reads<T>(&self, operation: impl FnOnce() -> T) -> T {
let _audit = crate::segment::reader::install_data_read_audit(Some(Arc::clone(
&self.counters.segment_bytes_read,
)));
operation()
}
#[must_use]
pub fn open_calls(&self) -> u64 {
self.counters.open_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn open_for_map_calls(&self) -> u64 {
self.counters.open_for_map_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn read_calls(&self) -> u64 {
self.counters.read_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn filtered_read_calls(&self) -> u64 {
self.counters.filtered_read_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn read_bytes(&self) -> u64 {
self.counters.read_bytes.load(Ordering::Relaxed)
}
#[must_use]
pub fn segment_bytes_read(&self) -> u64 {
self.counters.segment_bytes_read.load(Ordering::Relaxed)
}
#[must_use]
pub fn write_calls(&self) -> u64 {
self.counters.write_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn bytes_written(&self) -> u64 {
self.counters.bytes_written.load(Ordering::Relaxed)
}
#[must_use]
pub fn append_calls(&self) -> u64 {
self.counters.append_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn bytes_appended(&self) -> u64 {
self.counters.bytes_appended.load(Ordering::Relaxed)
}
#[must_use]
pub fn rename_calls(&self) -> u64 {
self.counters.rename_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn delete_calls(&self) -> u64 {
self.counters.delete_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn enumerate_calls(&self) -> u64 {
self.counters.enumerate_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn barrier_sync_calls(&self) -> u64 {
self.counters.barrier_sync_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn full_sync_calls(&self) -> u64 {
self.counters.full_sync_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn handle_barrier_sync_calls(&self) -> u64 {
self.counters
.handle_barrier_sync_calls
.load(Ordering::Relaxed)
}
#[must_use]
pub fn handle_full_sync_calls(&self) -> u64 {
self.counters.handle_full_sync_calls.load(Ordering::Relaxed)
}
pub fn reset(&self) {
self.counters.open_calls.store(0, Ordering::Relaxed);
self.counters.open_for_map_calls.store(0, Ordering::Relaxed);
self.counters.read_calls.store(0, Ordering::Relaxed);
self.counters
.filtered_read_calls
.store(0, Ordering::Relaxed);
self.counters.read_bytes.store(0, Ordering::Relaxed);
self.counters.segment_bytes_read.store(0, Ordering::Relaxed);
self.counters.write_calls.store(0, Ordering::Relaxed);
self.counters.bytes_written.store(0, Ordering::Relaxed);
self.counters.append_calls.store(0, Ordering::Relaxed);
self.counters.bytes_appended.store(0, Ordering::Relaxed);
self.counters.rename_calls.store(0, Ordering::Relaxed);
self.counters.delete_calls.store(0, Ordering::Relaxed);
self.counters.enumerate_calls.store(0, Ordering::Relaxed);
self.counters.barrier_sync_calls.store(0, Ordering::Relaxed);
self.counters.full_sync_calls.store(0, Ordering::Relaxed);
self.counters
.handle_barrier_sync_calls
.store(0, Ordering::Relaxed);
self.counters
.handle_full_sync_calls
.store(0, Ordering::Relaxed);
}
}
impl<V: Vfs> Vfs for CountingVfs<V> {
fn truncate(&self, path: &Path, length: u64) -> std::io::Result<()> {
self.inner.truncate(path, length)
}
fn create_new(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
self.inner.create_new(path, bytes)?;
self.counters.write_calls.fetch_add(1, Ordering::Relaxed);
self.counters
.bytes_written
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
Ok(())
}
fn segment_data_read_counter(&self) -> Option<Arc<AtomicU64>> {
Some(Arc::clone(&self.counters.segment_bytes_read))
}
fn create_directory(&self, path: &Path) -> std::io::Result<()> {
self.inner.create_directory(path)
}
fn remove_directory(&self, path: &Path) -> std::io::Result<()> {
self.inner.remove_directory(path)
}
fn ensure_directory(&self, path: &Path, create: bool) -> std::io::Result<bool> {
self.inner.ensure_directory(path, create)
}
fn open(&self, path: &Path) -> std::io::Result<u64> {
self.counters.open_calls.fetch_add(1, Ordering::Relaxed);
self.inner.open(path)
}
fn open_for_map(&self, path: &Path) -> std::io::Result<File> {
self.counters
.open_for_map_calls
.fetch_add(1, Ordering::Relaxed);
self.inner.open_for_map(path)
}
fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
let bytes = self.inner.read(path)?;
self.counters.read_calls.fetch_add(1, Ordering::Relaxed);
if self.read_path_filter.as_deref() == Some(path) {
self.counters
.filtered_read_calls
.fetch_add(1, Ordering::Relaxed);
}
self.counters
.read_bytes
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
if is_segment_path(path) {
self.counters
.segment_bytes_read
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
}
Ok(bytes)
}
fn read_range(&self, path: &Path, offset: u64, length: usize) -> std::io::Result<Vec<u8>> {
let bytes = self.inner.read_range(path, offset, length)?;
self.counters.read_calls.fetch_add(1, Ordering::Relaxed);
if self.read_path_filter.as_deref() == Some(path) {
self.counters
.filtered_read_calls
.fetch_add(1, Ordering::Relaxed);
}
self.counters
.read_bytes
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
if is_segment_path(path) {
self.counters
.segment_bytes_read
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
}
Ok(bytes)
}
fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
self.inner.write(path, bytes)?;
self.counters.write_calls.fetch_add(1, Ordering::Relaxed);
self.counters
.bytes_written
.fetch_add(bytes.len() as u64, Ordering::Relaxed);
Ok(())
}
fn open_append(&self, path: &Path) -> std::io::Result<Box<dyn VfsFile>> {
Ok(Box::new(CountingVfsFile {
inner: self.inner.open_append(path)?,
counters: Arc::clone(&self.counters),
}))
}
fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
self.inner.rename(from, to)?;
self.counters.rename_calls.fetch_add(1, Ordering::Relaxed);
Ok(())
}
fn sync(&self, path: &Path, kind: SyncKind) -> std::io::Result<()> {
self.inner.sync(path, kind)?;
match kind {
SyncKind::Barrier => &self.counters.barrier_sync_calls,
SyncKind::Full => &self.counters.full_sync_calls,
}
.fetch_add(1, Ordering::Relaxed);
Ok(())
}
fn list(&self, directory: &Path) -> std::io::Result<Vec<PathBuf>> {
self.inner.list(directory)
}
fn for_each_direct_child(
&self,
directory: &Path,
visitor: &mut dyn FnMut(&Path) -> std::io::Result<()>,
) -> std::io::Result<()> {
self.inner.for_each_direct_child(directory, visitor)?;
self.counters
.enumerate_calls
.fetch_add(1, Ordering::Relaxed);
Ok(())
}
fn delete(&self, path: &Path) -> std::io::Result<()> {
self.inner.delete(path)?;
self.counters.delete_calls.fetch_add(1, Ordering::Relaxed);
Ok(())
}
}
fn is_segment_path(path: &Path) -> bool {
path.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| name.starts_with("segment-") && name.ends_with(".zseg"))
}
#[cfg(not(windows))]
fn sync_path(path: &Path, kind: SyncKind) -> std::io::Result<()> {
sync_file(&File::open(path)?, kind)
}
#[cfg(windows)]
fn sync_path(path: &Path, _kind: SyncKind) -> std::io::Result<()> {
crate::sys::windows::sync_path(path)
}
#[cfg(any(target_os = "macos", target_os = "ios"))]
fn sync_file(file: &File, kind: SyncKind) -> std::io::Result<()> {
let result = match kind {
SyncKind::Barrier => crate::sys::darwin::barrier_fsync(file.as_raw_fd()),
SyncKind::Full => crate::sys::darwin::full_fsync(file.as_raw_fd()),
};
result.map_err(std::io::Error::other)
}
#[cfg(all(unix, not(any(target_os = "macos", target_os = "ios"))))]
fn sync_file(file: &File, kind: SyncKind) -> std::io::Result<()> {
let result = unsafe {
match kind {
SyncKind::Barrier => libc::fdatasync(file.as_raw_fd()),
SyncKind::Full => libc::fsync(file.as_raw_fd()),
}
};
if result == -1 {
Err(std::io::Error::last_os_error())
} else {
Ok(())
}
}
#[cfg(windows)]
fn sync_file(file: &File, _kind: SyncKind) -> std::io::Result<()> {
crate::sys::windows::flush_file(file)
}
#[cfg(not(any(unix, windows)))]
fn sync_file(_: &File, kind: SyncKind) -> std::io::Result<()> {
Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
format!("{kind:?} synchronization requires a supported file-descriptor platform"),
))
}
#[cfg(test)]
mod tests {
use std::path::Path;
use super::crash::MemoryVfs;
use super::{CountingVfs, SyncKind, Vfs};
#[test]
fn crash_support_api_gate_is_identical_in_debug_and_release() {
let source = include_str!("mod.rs");
let gate = source
.lines()
.zip(source.lines().skip(1))
.find_map(|(line, next)| (next == "pub mod crash;").then_some(line));
assert_eq!(gate, Some("#[cfg(any(test, feature = \"test-seams\"))]"));
}
#[test]
#[allow(clippy::expect_used)]
fn counting_vfs_counts_every_mutation_and_sync_kind_exactly() {
let inner = MemoryVfs::new();
inner
.insert("/source", b"abc".to_vec())
.expect("seed source");
let counting = CountingVfs::new(inner);
assert_eq!(counting.open(Path::new("/source")).expect("open"), 3);
assert_eq!(counting.read(Path::new("/source")).expect("read"), b"abc");
assert_eq!(
counting
.read_range(Path::new("/source"), 1, 2)
.expect("range"),
b"bc"
);
counting.write(Path::new("/write"), b"1234").expect("write");
let mut handle = counting
.open_append(Path::new("/append"))
.expect("open append");
handle.append(b"12").expect("first append");
handle.append(b"345").expect("second append");
handle.sync(SyncKind::Barrier).expect("handle barrier");
handle.sync(SyncKind::Full).expect("handle full");
counting
.sync(Path::new("/write"), SyncKind::Barrier)
.expect("path barrier");
counting
.sync(Path::new("/write"), SyncKind::Full)
.expect("path full");
counting
.rename(Path::new("/write"), Path::new("/renamed"))
.expect("rename");
counting.delete(Path::new("/renamed")).expect("delete");
assert_eq!(
[
counting.open_calls(),
counting.read_calls(),
counting.read_bytes(),
counting.write_calls(),
counting.bytes_written(),
counting.append_calls(),
counting.bytes_appended(),
counting.rename_calls(),
counting.delete_calls(),
counting.barrier_sync_calls(),
counting.full_sync_calls(),
counting.handle_barrier_sync_calls(),
counting.handle_full_sync_calls(),
],
[1, 2, 5, 1, 4, 2, 5, 1, 1, 1, 1, 1, 1],
"CountingVfs must expose every exact VFS and open-handle cost"
);
counting.reset();
assert_eq!(
[
counting.open_calls(),
counting.read_calls(),
counting.read_bytes(),
counting.write_calls(),
counting.bytes_written(),
counting.append_calls(),
counting.bytes_appended(),
counting.rename_calls(),
counting.delete_calls(),
counting.barrier_sync_calls(),
counting.full_sync_calls(),
counting.handle_barrier_sync_calls(),
counting.handle_full_sync_calls(),
],
[0; 13],
"reset must zero decorator and shared-handle counters"
);
}
}
#[cfg(feature = "test-seams")]
pub mod file_test_support;