use std::io::{Read, Seek, SeekFrom, Write};
use std::sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
};
use vfsi_sync::api::ReadOptions;
use vfsi_sync::{
Backend, Capabilities, DirEntry, DirPageCursor, FileSystem, FsClient, MetadataQuery, OpenFlags,
OpenOp, ReadDirOptions, ReadIntoResult, ReadOp, ReadResult, SetAttributes, VfAttrs, VfError,
VfFile, VfOffset, VfResult, Vfsi, VfsiExt, WriteOpRef, WriteResult,
};
#[test]
fn client_policy_applies_to_scalar_vector_and_into_reads_before_dispatch() {
let backend = ScalarOnly {
data: b"abcdef".to_vec(),
..ScalarOnly::default()
};
let read_calls = Arc::clone(&backend.read_calls);
let client = FsClient::new(backend).with_limits(vfsi_sync::ResourceLimits {
max_read_bytes: 3,
stream_chunk_bytes: 2,
..vfsi_sync::ResourceLimits::default()
});
assert_eq!(client.clone().limits(), client.limits());
assert_eq!(client.capabilities().unwrap(), Capabilities::empty());
assert_eq!(
client.read("/file").unwrap_err().kind(),
std::io::ErrorKind::FileTooLarge
);
assert_eq!(
client
.read_with_options(
"/file",
ReadOptions::new().max_total_bytes(std::num::NonZeroUsize::new(6))
)
.unwrap(),
b"abcdef"
);
let file = client.open("/file").unwrap();
let before = read_calls.load(Ordering::SeqCst);
assert!(client.vread_native(&[file.read_request_at(0, 4)]).is_err());
let mut buffer = [0; 4];
assert!(
client
.vread_into_native(&mut [file.read_request_at_into(0, &mut buffer)])
.is_err()
);
assert_eq!(read_calls.load(Ordering::SeqCst), before);
let result = client
.vread_into_with_limit_native(&mut [file.read_request_at_into(0, &mut buffer)], 4)
.unwrap();
assert_eq!(result[0].read, 4);
assert_eq!(&buffer, b"abcd");
buffer.fill(0xff);
let before = read_calls.load(Ordering::SeqCst);
assert!(
client
.vread_into_with_limit_native(&mut [file.read_request_at_into(0, &mut buffer)], 3)
.is_err()
);
assert_eq!(read_calls.load(Ordering::SeqCst), before);
assert_eq!(buffer, [0xff; 4]);
let mut offsets = Vec::new();
let result = client
.read_stream("/file", |offset, bytes| {
offsets.push((offset, bytes.len()));
Ok(offset == 0)
})
.unwrap();
assert_eq!(offsets, [(0, 2), (2, 2)]);
assert_eq!(
result,
vfsi_sync::StreamCompletion::Stopped { next_offset: 4 }
);
assert_eq!(
client.read_stream("/file", |_, _| Ok(true)).unwrap(),
vfsi_sync::StreamCompletion::Complete
);
}
#[test]
fn scalar_read_uses_bounded_whole_file_api_without_opening_a_handle() {
for read_failure in [false, true] {
let client = FsClient::new(ScalarOnly {
data: b"data".to_vec(),
close_failures_remaining: 1,
read_failure,
..ScalarOnly::default()
});
let result = client.read("/file");
if read_failure {
let error = result.unwrap_err();
assert!(!error.is_transport());
assert_eq!(error.operation(), Some("read"));
} else {
assert_eq!(result.unwrap(), b"data");
}
let backend = client.into_inner().unwrap();
assert_eq!(backend.close_calls, 0);
assert!(!backend.open);
}
}
type Notifications = Arc<Mutex<Vec<Box<dyn FnOnce() + Send>>>>;
#[derive(Default)]
struct ScalarOnly {
stats_calls: Arc<AtomicUsize>,
stats_result_count: Option<usize>,
stats_error_index: Option<usize>,
notifications: Notifications,
blocked_read: Option<(Arc<std::sync::Barrier>, Arc<std::sync::Barrier>)>,
close_observed: Arc<AtomicUsize>,
data: Vec<u8>,
cursor: u64,
open: bool,
oversized_read: bool,
oversized_write_count: bool,
read_failure: bool,
transport_failure: bool,
vector_result_limit: Arc<Mutex<Option<usize>>>,
wrong_result_file: bool,
wrong_result_offset: bool,
close_failures_remaining: usize,
close_calls: usize,
direct_into_only: bool,
into_calls: Arc<AtomicUsize>,
read_calls: Arc<AtomicUsize>,
write_calls: Arc<AtomicUsize>,
max_write_once: Option<usize>,
directory_entries: usize,
directory_page_sizes: Arc<Mutex<Vec<usize>>>,
directory_fields: Arc<Mutex<Vec<vfsi_sync::AttrMask>>>,
}
#[test]
fn opened_file_read_to_end_has_explicit_limits_and_cursor_semantics() {
for limit in [0, 3, 6, 7] {
let client = FsClient::new(ScalarOnly {
data: b"abcdef".to_vec(),
..Default::default()
})
.with_limits(vfsi_sync::ResourceLimits {
max_read_bytes: 1,
..Default::default()
});
let mut file = client.open("/file").unwrap();
let result = file.read_to_end_with_limit(limit);
if limit < 6 {
let error = result.unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::FileTooLarge);
assert_eq!(error.operation(), Some("read_to_end_with_limit"));
assert_eq!(
file.seek_native(SeekFrom::Current(0)).unwrap(),
(limit + 1) as u64
);
} else {
assert_eq!(result.unwrap(), b"abcdef");
assert_eq!(file.seek_native(SeekFrom::Current(0)).unwrap(), 6);
}
file.seek_native(SeekFrom::Start(3)).unwrap();
assert_eq!(file.read_to_end_with_limit(3).unwrap(), b"def");
assert_eq!(file.read_to_end_with_limit(0).unwrap(), b"");
file.try_close().unwrap();
assert_eq!(
file.read_to_end_with_limit(0).unwrap_err().err_no(),
libc::EBADF as u32
);
}
}
impl Backend for ScalarOnly {
fn vopen_impl(&mut self, requests: &[OpenOp]) -> VfResult<Vec<VfFile>> {
if self.transport_failure {
return Err(VfError::transport(None, "reply lost"));
}
let limit = self
.vector_result_limit
.lock()
.expect("vector limit poisoned")
.unwrap_or(usize::MAX);
requests
.iter()
.take(limit)
.map(|request| self.open_impl(request))
.collect()
}
fn vclose_impl(&mut self, files: &[VfFile]) -> VfResult<()> {
for file in files {
self.close_impl(file)?;
}
Ok(())
}
fn listdir_page_impl(
&mut self,
dir: &std::path::Path,
masks: vfsi_sync::AttrMask,
cursor: Option<DirPageCursor>,
page_size: usize,
max_entries: usize,
) -> VfResult<(Vec<VfAttrs>, Option<DirPageCursor>)> {
self.directory_fields.lock().unwrap().push(masks);
self.directory_page_sizes.lock().unwrap().push(page_size);
let start = cursor
.map(|cursor| cursor.into_state::<usize>())
.transpose()?
.unwrap_or(0);
let count = if max_entries == 0 {
self.directory_entries
} else {
self.directory_entries
.min(start.saturating_add(max_entries))
};
let end = start.saturating_add(page_size).min(count);
let entries = (start..end)
.map(|index| VfAttrs {
file: VfFile::from_os_path(&dir.join(format!("item-{index:04}"))),
masks,
mode: 0o100644,
size: index as u64,
..VfAttrs::default()
})
.collect();
Ok((entries, (end < count).then(|| DirPageCursor::new(end))))
}
fn vread_impl(&mut self, requests: &[ReadOp]) -> VfResult<Vec<ReadResult>> {
let limit = self
.vector_result_limit
.lock()
.expect("vector limit poisoned")
.unwrap_or(usize::MAX);
let mut results: Vec<_> = requests
.iter()
.take(limit)
.map(|request| self.read_impl(request))
.collect::<VfResult<_>>()?;
if let Some(result) = results.first_mut() {
if self.wrong_result_file {
result.file = VfFile::from_fd(999);
}
if self.wrong_result_offset {
result.offset = result.offset.saturating_add(1);
}
}
Ok(results)
}
fn vread_into_impl(
&mut self,
requests: &[ReadOp],
buffers: &mut [&mut [u8]],
) -> VfResult<Vec<ReadIntoResult>> {
if requests.len() != buffers.len() {
return Err(VfError::client(0, libc::EINVAL as u32));
}
let limit = self
.vector_result_limit
.lock()
.expect("vector limit poisoned")
.unwrap_or(usize::MAX);
let mut results: Vec<ReadIntoResult> = requests
.iter()
.zip(buffers.iter_mut())
.take(limit)
.enumerate()
.map(|(index, (request, buffer))| {
self.read_into_impl(request, buffer)
.map_err(|error| error.with_index(index))
})
.collect::<VfResult<_>>()?;
if let Some(result) = results.first_mut() {
if self.wrong_result_file {
result.file = VfFile::from_fd(999);
}
if self.wrong_result_offset {
result.offset = result.offset.saturating_add(1);
}
}
Ok(results)
}
fn vwrite_impl(&mut self, requests: &[WriteOpRef<'_>]) -> VfResult<Vec<WriteResult>> {
let limit = self
.vector_result_limit
.lock()
.expect("vector limit poisoned")
.unwrap_or(usize::MAX);
let mut results: Vec<_> = requests
.iter()
.take(limit)
.map(|request| self.write_impl(*request))
.collect::<VfResult<_>>()?;
if let Some(result) = results.first_mut() {
if self.wrong_result_file {
result.file = VfFile::from_fd(999);
}
if self.wrong_result_offset {
result.offset = result.offset.saturating_add(1);
}
}
Ok(results)
}
fn read_dir_page_with_fields_impl(
&mut self,
path: &std::path::Path,
fields: vfsi_sync::AttrMask,
cursor: Option<DirPageCursor>,
page_size: usize,
max_entries: usize,
) -> VfResult<(Vec<DirEntry>, Option<DirPageCursor>)> {
self.directory_fields.lock().unwrap().push(fields);
self.read_dir_page_impl(path, cursor, page_size, max_entries)
}
fn create_dir_impl(&mut self, _: &std::path::Path, _: u32) -> VfResult<()> {
Ok(())
}
fn read_dir_impl(&mut self, _: &std::path::Path, _: ReadDirOptions) -> VfResult<Vec<DirEntry>> {
unreachable!("the visitor must use paged enumeration")
}
fn read_dir_page_impl(
&mut self,
_: &std::path::Path,
cursor: Option<DirPageCursor>,
page_size: usize,
max_entries: usize,
) -> VfResult<(Vec<DirEntry>, Option<DirPageCursor>)> {
self.directory_page_sizes.lock().unwrap().push(page_size);
let start = cursor
.map(|cursor| cursor.into_state::<usize>())
.transpose()?
.unwrap_or(0);
let ceiling = if max_entries == 0 {
self.directory_entries
} else {
self.directory_entries.min(max_entries)
};
let end = start.saturating_add(page_size).min(ceiling);
let entries = (start..end)
.map(|index| {
DirEntry::new(
format!("/tree/item-{index:04}").into(),
vfsi_core::metadata_from_attrs(VfAttrs::default()),
)
})
.collect();
let next = (end < ceiling).then(|| DirPageCursor::new(end));
Ok((entries, next))
}
}
#[derive(Default)]
struct HandleOnly {
scalar: ScalarOnly,
}
#[derive(Default)]
struct DefaultBackend {
scalar: ScalarOnly,
vector_reads: usize,
}
#[derive(Default)]
struct PagedBackend {
scalar: ScalarOnly,
page_calls: Arc<Mutex<Vec<(vfsi_sync::AttrMask, usize, usize)>>>,
}
macro_rules! handle_contract {
($backend:ty) => {
impl FileSystem for $backend {
fn open_impl(&mut self, request: &OpenOp) -> VfResult<VfFile> {
self.scalar.open_impl(request)
}
fn open_path_impl(
&mut self,
base: vfsi_sync::VfPathBase,
path: &std::path::Path,
flags: i32,
mode: u32,
) -> VfResult<VfFile> {
let _ = (base, flags, mode);
self.open_impl(&OpenOp::new(path, OpenFlags::READ))
}
fn close_impl(&mut self, file: &VfFile) -> VfResult<()> {
self.scalar.close_impl(file)
}
fn sync_data(&mut self, file: &VfFile) -> VfResult<()> {
self.scalar.sync_data(file)
}
fn read_impl(&mut self, request: &ReadOp) -> VfResult<ReadResult> {
self.scalar.read_impl(request)
}
fn write_impl(&mut self, request: WriteOpRef<'_>) -> VfResult<WriteResult> {
self.scalar.write_impl(request)
}
fn seek_impl(&mut self, file: &VfFile, position: SeekFrom) -> VfResult<u64> {
self.scalar.seek_impl(file, position)
}
fn metadata_impl(&mut self, query: MetadataQuery) -> VfResult<VfAttrs> {
self.scalar.metadata_impl(query)
}
fn set_attributes_impl(&mut self, update: SetAttributes) -> VfResult<()> {
self.scalar.set_attributes_impl(update)
}
}
};
}
handle_contract!(HandleOnly);
handle_contract!(DefaultBackend);
handle_contract!(PagedBackend);
impl Backend for PagedBackend {
fn vread_impl(&mut self, _: &[ReadOp]) -> VfResult<Vec<ReadResult>> {
panic!("directory paging must not issue reads")
}
fn listdir_page_impl(
&mut self,
dir: &std::path::Path,
masks: vfsi_sync::AttrMask,
cursor: Option<DirPageCursor>,
page_size: usize,
max_entries: usize,
) -> VfResult<(Vec<VfAttrs>, Option<DirPageCursor>)> {
self.page_calls
.lock()
.unwrap()
.push((masks, page_size, max_entries));
let start = cursor
.map(|cursor| cursor.into_state::<usize>())
.transpose()?
.unwrap_or(0);
let count = if max_entries == 0 {
3
} else {
3.min(max_entries)
};
let end = start.saturating_add(page_size).min(count);
let entries = (start..end)
.map(|index| VfAttrs {
file: VfFile::from_os_path(&dir.join(format!("item-{index}"))),
masks,
mode: 0o100644,
size: index as u64,
..VfAttrs::default()
})
.collect();
Ok((entries, (end < count).then(|| DirPageCursor::new(end))))
}
}
#[test]
fn minimal_backend_directory_defaults_return_unsupported() {
let mut concrete = DefaultBackend::default();
let backend: &mut dyn Backend = &mut concrete;
let dir = std::path::Path::new("/tree");
assert_eq!(
backend
.read_dir_page_impl(dir, None, 1, 4)
.err()
.expect("paging is unsupported")
.kind(),
std::io::ErrorKind::Unsupported,
);
assert_eq!(
backend
.read_dir_page_with_fields_impl(dir, vfsi_sync::AttrMask::SIZE, None, 1, 4)
.err()
.expect("paging is unsupported")
.kind(),
std::io::ErrorKind::Unsupported,
);
let client = FsClient::new(concrete);
assert_eq!(
client
.visit_dir(dir, |_| panic!(
"unsupported listing must not invoke the visitor"
))
.unwrap_err()
.kind(),
std::io::ErrorKind::Unsupported,
);
}
#[test]
fn default_directory_pages_preserve_cursor_fields_and_bounds() {
let mut concrete = PagedBackend::default();
let calls = concrete.page_calls.clone();
let backend: &mut dyn Backend = &mut concrete;
let dir = std::path::Path::new("/tree");
let (first, next) = backend.read_dir_page_impl(dir, None, 1, 3).unwrap();
assert_eq!(first.len(), 1);
assert_eq!(first[0].path(), dir.join("item-0"));
let (rest, next) = backend
.read_dir_page_with_fields_impl(dir, vfsi_sync::AttrMask::SIZE, next, 2, 3)
.unwrap();
assert_eq!(
rest.iter().map(|entry| entry.path()).collect::<Vec<_>>(),
[dir.join("item-1"), dir.join("item-2")]
);
assert!(next.is_none());
let calls = calls.lock().unwrap();
assert_eq!(calls.len(), 2);
assert!(
calls[0]
.0
.contains(vfsi_sync::AttrMask::MODE | vfsi_sync::AttrMask::SIZE)
);
assert_eq!((calls[0].1, calls[0].2), (1, 3));
assert_eq!(
calls[1],
(vfsi_sync::AttrMask::MODE | vfsi_sync::AttrMask::SIZE, 2, 3)
);
}
#[test]
fn default_directory_visitor_uses_native_pages_without_materializing_a_listing() {
let concrete = PagedBackend::default();
let calls = concrete.page_calls.clone();
let client = FsClient::new(concrete);
let mut paths = Vec::new();
let completion = client
.visit_dir_with_options(
"/tree",
vfsi_sync::ListDirOptions::new()
.fields(vfsi_sync::AttrMask::SIZE)
.max_entries(3),
|entry| {
paths.push(entry.path().to_path_buf());
Ok(std::ops::ControlFlow::Continue(()))
},
)
.unwrap();
assert_eq!(completion, vfsi_sync::TraversalCompletion::Complete);
assert_eq!(
paths,
(0..3)
.map(|i| format!("/tree/item-{i}").into())
.collect::<Vec<std::path::PathBuf>>()
);
assert_eq!(
*calls.lock().unwrap(),
[
(vfsi_sync::AttrMask::MODE | vfsi_sync::AttrMask::SIZE, 1, 4),
(vfsi_sync::AttrMask::MODE | vfsi_sync::AttrMask::SIZE, 3, 3),
]
);
}
#[derive(Default)]
struct WorkflowOverride {
scalar: ScalarOnly,
calls: Vec<&'static str>,
}
handle_contract!(WorkflowOverride);
impl Backend for WorkflowOverride {
fn vread_impl(&mut self, _: &[ReadOp]) -> VfResult<Vec<ReadResult>> {
panic!("specialized workflows must not fall back to generic vector reads")
}
fn read_dir_page_with_fields_impl(
&mut self,
_: &std::path::Path,
fields: vfsi_sync::AttrMask,
_: Option<DirPageCursor>,
page_size: usize,
max_entries: usize,
) -> VfResult<(Vec<DirEntry>, Option<DirPageCursor>)> {
assert!(fields.contains(vfsi_sync::AttrMask::MODE | vfsi_sync::AttrMask::SIZE));
assert_eq!((page_size, max_entries), (1, 4));
self.calls.push("directory");
Ok((Vec::new(), None))
}
fn vstream_impl(
&mut self,
_: &[VfFile],
_: usize,
_: usize,
callback: &mut vfsi_sync::ReadStreamCallback<'_>,
) -> vfsi_sync::VfRes {
self.calls.push("stream");
callback(0, 0, b"override", true);
Ok(())
}
fn walk_with_options_impl(
&mut self,
_: &std::path::Path,
_: vfsi_sync::AttrMask,
_: vfsi_sync::WalkOptions,
_: &mut dyn FnMut(&std::path::Path, &mut Vec<VfAttrs>),
) -> VfResult<Vec<vfsi_sync::WalkEntry>> {
self.calls.push("walk");
Ok(Vec::new())
}
fn remove_paths_with_options_impl(
&mut self,
_: &[&std::path::Path],
_: bool,
_: vfsi_sync::RemoveOptions,
) -> vfsi_sync::VfRes {
self.calls.push("remove");
Ok(())
}
fn vcopy_impl(
&mut self,
_: &[vfsi_sync::ExtentPair],
_: vfsi_sync::CopyOption,
) -> vfsi_sync::VfRes {
self.calls.push("copy");
Ok(())
}
}
#[test]
fn shared_workflows_keep_dynamic_backend_overrides_reachable() {
let mut concrete = WorkflowOverride::default();
let backend: &mut dyn Backend = &mut concrete;
let files = [VfFile::from_fd(1)];
assert_eq!(
backend.vread_all_impl(&files).unwrap(),
[b"override".to_vec()]
);
assert!(
backend
.walk_impl(
std::path::Path::new("/tree"),
vfsi_sync::AttrMask::default(),
&mut |_, _| {},
)
.unwrap()
.is_empty()
);
backend
.remove_impl(std::path::Path::new("/tree"), true)
.unwrap();
backend
.vcopy_impl(
&[vfsi_sync::ExtentPair::from_os_paths(
std::path::Path::new("/a"),
0,
std::path::Path::new("/b"),
0,
None,
)],
vfsi_sync::CopyOption::default(),
)
.unwrap();
let (entries, next) = backend
.read_dir_page_impl(std::path::Path::new("/tree"), None, 1, 4)
.unwrap();
assert!(entries.is_empty());
assert!(next.is_none());
assert_eq!(
concrete.calls,
["stream", "walk", "remove", "copy", "directory"]
);
}
impl Backend for DefaultBackend {
fn vread_impl(&mut self, requests: &[ReadOp]) -> VfResult<Vec<ReadResult>> {
self.vector_reads += 1;
requests
.iter()
.map(|request| self.read_impl(request))
.collect()
}
}
#[test]
fn minimal_backend_defaults_are_object_safe_bounded_and_terminate() {
let mut concrete = DefaultBackend {
scalar: ScalarOnly {
data: b"abcdef".to_vec(),
..Default::default()
},
..Default::default()
};
let backend: &mut dyn Backend = &mut concrete;
let files = backend
.vopen_impl(&[OpenOp::new("/file", OpenFlags::READ)])
.unwrap();
let data = backend
.vread_all_with_options_impl(&files, vfsi_sync::ReadAllOptions::new().max_total_bytes(6))
.unwrap();
assert_eq!(data, [b"abcdef".to_vec()]);
assert_eq!(
backend
.vread_all_with_options_impl(
&files,
vfsi_sync::ReadAllOptions::new().max_total_bytes(3),
)
.unwrap_err()
.kind(),
std::io::ErrorKind::FileTooLarge
);
let mut received = Vec::new();
backend
.vstream_impl(&files, 2, 4, &mut |index, offset, bytes, eof| {
assert_eq!(index, 0);
assert_eq!(offset as usize, received.len());
assert!(bytes.len() <= 2);
received.extend_from_slice(bytes);
assert_eq!(eof, received.len() == 6);
true
})
.unwrap();
assert_eq!(received, b"abcdef");
let mut buffer = [0; 3];
let result = backend
.vread_into_impl(&[ReadOp::at(files[0].clone(), 2, 3)], &mut [&mut buffer])
.unwrap();
assert_eq!(buffer, *b"cde");
assert_eq!(result[0].read, 3);
let paths = [(VfFile::from_path("/source"), VfFile::from_path("/target"))];
assert_eq!(
backend
.rename_impl(
std::path::Path::new("/source"),
std::path::Path::new("/target")
)
.unwrap_err()
.kind(),
std::io::ErrorKind::Unsupported
);
assert_eq!(
backend
.vrename_with_options_impl(&paths, vfsi_core::api::RenameOptions::NoReplace)
.unwrap_err()
.kind(),
std::io::ErrorKind::Unsupported
);
assert_eq!(
backend
.read_link_impl(std::path::Path::new("/link"))
.unwrap_err()
.kind(),
std::io::ErrorKind::Unsupported
);
assert_eq!(
backend
.vwrite_impl(&[WriteOpRef::new(&files[0], VfOffset::At(0), b"x")])
.unwrap_err()
.kind(),
std::io::ErrorKind::Unsupported
);
backend.vclose_impl(&files).unwrap();
assert!(!concrete.scalar.open);
assert_eq!(concrete.scalar.close_calls, 1);
assert!(concrete.vector_reads >= 5);
}
impl FileSystem for ScalarOnly {
fn vstatfs_impl(&mut self, files: &[VfFile]) -> VfResult<Vec<vfsi_sync::FilesystemStats>> {
self.stats_calls.fetch_add(1, Ordering::SeqCst);
if let Some(index) = self.stats_error_index {
return Err(VfError::client(index, libc::ENOENT as u32));
}
Ok(vec![
vfsi_sync::FilesystemStats::default();
self.stats_result_count.unwrap_or(files.len())
])
}
fn take_notifications(&mut self) -> Vec<Box<dyn FnOnce() + Send>> {
std::mem::take(&mut *self.notifications.lock().unwrap())
}
fn capabilities(&self) -> Capabilities {
Capabilities::empty()
}
fn open_impl(&mut self, _: &OpenOp) -> VfResult<VfFile> {
self.open = true;
Ok(VfFile::from_fd(1))
}
fn close_impl(&mut self, _: &VfFile) -> VfResult<()> {
self.close_calls += 1;
self.close_observed.fetch_add(1, Ordering::SeqCst);
if self.close_failures_remaining != 0 {
self.close_failures_remaining -= 1;
return Err(VfError::transport(None, "injected close failure"));
}
self.open = false;
Ok(())
}
fn sync_data(&mut self, _: &VfFile) -> VfResult<()> {
Ok(())
}
fn sync_all(&mut self, _: &VfFile) -> VfResult<()> {
Ok(())
}
fn read_impl(&mut self, request: &ReadOp) -> VfResult<ReadResult> {
if let Some((entered, release)) = self.blocked_read.take() {
entered.wait();
release.wait();
}
self.read_calls.fetch_add(1, Ordering::SeqCst);
assert!(!self.direct_into_only, "owned read path must not run");
if self.read_failure {
return Err(VfError::failure(0, libc::EACCES as u32));
}
if self.oversized_read {
return Ok(ReadResult {
file: request.file.clone(),
offset: 0,
eof: false,
data: vec![0; request.length + 1],
});
}
let offset = match request.offset {
VfOffset::At(value) => value,
VfOffset::Cur => self.cursor,
VfOffset::End => self.data.len() as u64,
_ => 0,
};
let start = offset as usize;
let end = start.saturating_add(request.length).min(self.data.len());
let data = self.data.get(start..end).unwrap_or_default().to_vec();
self.cursor = offset + data.len() as u64;
Ok(ReadResult {
file: request.file.clone(),
offset,
eof: end == self.data.len(),
data,
})
}
fn read_into_impl(&mut self, request: &ReadOp, buffer: &mut [u8]) -> VfResult<ReadIntoResult> {
if !self.direct_into_only {
let result = self.read_impl(request)?;
if result.data.len() > buffer.len() {
return Err(VfError::client(0, vfsi_sync::ERR_IO));
}
buffer[..result.data.len()].copy_from_slice(&result.data);
return Ok(ReadIntoResult {
file: result.file,
offset: result.offset,
read: result.data.len(),
eof: result.eof,
});
}
if request.length != buffer.len() {
return Err(VfError::client(0, libc::EINVAL as u32));
}
self.into_calls.fetch_add(1, Ordering::SeqCst);
let offset = match request.offset {
VfOffset::At(value) => value,
VfOffset::Cur => self.cursor,
VfOffset::End => self.data.len() as u64,
_ => 0,
};
let start = offset as usize;
let end = start.saturating_add(buffer.len()).min(self.data.len());
let data = self.data.get(start..end).unwrap_or_default();
buffer[..data.len()].copy_from_slice(data);
self.cursor = offset + data.len() as u64;
Ok(ReadIntoResult {
file: request.file.clone(),
offset,
read: data.len(),
eof: end == self.data.len(),
})
}
fn write_impl(&mut self, request: WriteOpRef<'_>) -> VfResult<WriteResult> {
self.write_calls.fetch_add(1, Ordering::SeqCst);
if self.oversized_write_count {
return Ok(WriteResult {
file: request.file.clone(),
offset: 0,
written: request.data.len() + 1,
stable: true,
});
}
let offset = match request.offset {
VfOffset::At(value) => value,
VfOffset::Cur => self.cursor,
VfOffset::End => self.data.len() as u64,
_ => 0,
};
let start = offset as usize;
let length = request
.data
.len()
.min(self.max_write_once.unwrap_or(usize::MAX));
self.data.resize(self.data.len().max(start + length), 0);
self.data[start..start + length].copy_from_slice(&request.data[..length]);
self.cursor = offset + length as u64;
Ok(WriteResult {
file: request.file.clone(),
offset,
written: length,
stable: true,
})
}
fn seek_impl(&mut self, _: &VfFile, position: SeekFrom) -> VfResult<u64> {
let next = match position {
SeekFrom::Start(value) => value as i128,
SeekFrom::Current(value) => self.cursor as i128 + value as i128,
SeekFrom::End(value) => self.data.len() as i128 + value as i128,
};
if next < 0 || next > u64::MAX as i128 {
return Err(VfError::failure(0, libc::EINVAL as u32));
}
self.cursor = next as u64;
Ok(self.cursor)
}
fn metadata_impl(&mut self, _: MetadataQuery) -> VfResult<vfsi_sync::VfAttrs> {
Err(VfError::unsupported(0))
}
fn set_attributes_impl(&mut self, _: SetAttributes) -> VfResult<()> {
Err(VfError::unsupported(0))
}
}
#[test]
fn owned_client_accepts_a_scalar_only_backend() {
let client = FsClient::new(HandleOnly::default());
let mut options = client.open_options();
options.read(true).write(true).create(true);
let mut file = options.open("/file").unwrap();
file.write_all(b"scalar").unwrap();
file.seek(SeekFrom::Start(0)).unwrap();
let mut output = String::new();
file.read_to_string(&mut output).unwrap();
assert_eq!(output, "scalar");
file.close().unwrap();
}
#[test]
fn owned_file_rejects_backend_results_that_violate_io_contracts() {
let client = FsClient::new(ScalarOnly {
oversized_read: true,
..ScalarOnly::default()
});
let mut file = client
.open_with(OpenOp::new("/file", OpenFlags::READ))
.unwrap();
let error = file.read_native(&mut [0; 4]).unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_IO);
assert_eq!(error.operation(), Some("read"));
assert_eq!(error.path(), Some(std::path::Path::new("/file")));
let client = FsClient::new(ScalarOnly {
oversized_write_count: true,
..ScalarOnly::default()
});
let mut file = client
.open_with(OpenOp::new("/file", OpenFlags::WRITE))
.unwrap();
let error = file.write_native(b"data").unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_IO);
assert_eq!(error.operation(), Some("write"));
assert_eq!(error.path(), Some(std::path::Path::new("/file")));
}
#[test]
fn native_file_errors_retain_operation_and_path_context() {
let client = FsClient::new(ScalarOnly {
read_failure: true,
..ScalarOnly::default()
});
let file = client.open("/important").unwrap();
let error = file.read_at(&mut [0; 1], 0).unwrap_err();
assert_eq!(error.err_no(), libc::EACCES as u32);
assert_eq!(error.operation(), Some("read"));
assert_eq!(error.path(), Some(std::path::Path::new("/important")));
}
#[test]
fn convenience_string_errors_retain_operation_and_path_context() {
let client = FsClient::new(ScalarOnly {
data: vec![0xff],
..ScalarOnly::default()
});
let error = client.read_to_string("/not-utf8").unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_INVAL);
assert_eq!(error.operation(), Some("read_to_string"));
assert_eq!(error.path(), Some(std::path::Path::new("/not-utf8")));
}
#[test]
fn allocating_whole_file_reads_enforce_a_configurable_limit() {
let client = FsClient::new(ScalarOnly {
data: b"four".to_vec(),
..ScalarOnly::default()
});
let error = client
.read_with_options(
"/file",
ReadOptions::new().max_total_bytes(std::num::NonZeroUsize::new(3)),
)
.unwrap_err();
assert_eq!(error.err_no(), libc::EFBIG as u32);
assert_eq!(error.operation(), Some("read"));
assert_eq!(error.path(), Some(std::path::Path::new("/file")));
let client = FsClient::new(ScalarOnly {
data: b"four".to_vec(),
..ScalarOnly::default()
});
assert_eq!(
client
.read_with_options(
"/file",
ReadOptions::new().max_total_bytes(std::num::NonZeroUsize::new(4))
)
.unwrap(),
b"four"
);
let client = FsClient::new(ScalarOnly {
data: b"utf8".to_vec(),
..ScalarOnly::default()
});
assert_eq!(
client
.read_to_string_with_options(
"/file",
ReadOptions::new().max_total_bytes(std::num::NonZeroUsize::new(4))
)
.unwrap(),
"utf8"
);
}
#[test]
fn vector_transport_failure_does_not_invent_request_zero_context() {
let client = FsClient::new(ScalarOnly {
transport_failure: true,
..ScalarOnly::default()
});
let error = client
.vopen(&[
OpenOp::new("/first", OpenFlags::READ),
OpenOp::new("/second", OpenFlags::READ),
])
.unwrap_err();
assert!(error.is_transport());
assert_eq!(error.index(), None);
assert_eq!(error.path(), None);
}
#[test]
fn openv_rejects_wrong_result_count_and_cleans_returned_handles() {
let client = FsClient::new(ScalarOnly {
vector_result_limit: Arc::new(Mutex::new(Some(1))),
..ScalarOnly::default()
});
let error = client
.vopen(&[
OpenOp::new("/first", OpenFlags::READ),
OpenOp::new("/second", OpenFlags::READ),
])
.unwrap_err();
assert!(error.is_transport());
assert_eq!(error.index(), None);
assert!(error.to_string().contains("1 results for 2 requests"));
assert!(!client.into_inner().unwrap().open);
}
#[test]
fn readv_and_writev_reject_wrong_result_counts() {
let backend = ScalarOnly::default();
let vector_result_limit = Arc::clone(&backend.vector_result_limit);
let client = FsClient::new(backend);
let files = client
.vopen(&[
OpenOp::new("/first", OpenFlags::READ | OpenFlags::WRITE),
OpenOp::new("/second", OpenFlags::READ | OpenFlags::WRITE),
])
.unwrap();
*vector_result_limit.lock().unwrap() = Some(1);
let read_error = client
.vread_native(&[
files[0].read_request_at(0, 1),
files[1].read_request_at(0, 1),
])
.unwrap_err();
assert!(read_error.is_transport());
assert_eq!(read_error.index(), None);
let write_error = client
.vwrite_native(&[
files[0].write_request_at(0, b"a"),
files[1].write_request_at(0, b"b"),
])
.unwrap_err();
assert!(write_error.is_transport());
assert_eq!(write_error.index(), None);
client.vclose_owned(files).unwrap();
}
#[test]
fn owned_vector_reads_reject_oversized_batches_before_backend_io() {
let read_calls = Arc::new(AtomicUsize::new(0));
let client = FsClient::new(ScalarOnly {
data: b"abcdef".to_vec(),
read_calls: Arc::clone(&read_calls),
..ScalarOnly::default()
});
let file = client.open("/file").unwrap();
let error = client
.vread_with_limit_native(&[file.read_request_at(0, 3), file.read_request_at(3, 3)], 5)
.unwrap_err();
assert_eq!(error.index(), Some(1));
assert_eq!(error.err_no(), libc::EFBIG as u32);
assert_eq!(error.path(), Some(std::path::Path::new("/file")));
assert_eq!(read_calls.load(Ordering::SeqCst), 0);
let error = client
.vread_native(&[file.read_request_at(0, vfsi_sync::DEFAULT_READV_MAX_TOTAL_BYTES + 1)])
.unwrap_err();
assert_eq!(error.err_no(), libc::EFBIG as u32);
assert_eq!(read_calls.load(Ordering::SeqCst), 0);
let results = client
.vread_with_limit_native(&[file.read_request_at(0, 3)], 3)
.unwrap();
assert_eq!(results[0].data, b"abc");
}
#[test]
fn write_allv_retries_short_writes_in_vector_waves() {
let write_calls = Arc::new(AtomicUsize::new(0));
let client = FsClient::new(ScalarOnly {
max_write_once: Some(2),
write_calls: Arc::clone(&write_calls),
..ScalarOnly::default()
});
let file = client
.open_with(OpenOp::new("/file", OpenFlags::WRITE))
.unwrap();
let results = client
.vwrite_all_native(&[
file.write_request_at(0, b"abcde"),
file.write_request_at(10, b"VWXYZ"),
])
.unwrap();
assert_eq!(
results
.iter()
.map(|result| result.written)
.collect::<Vec<_>>(),
[5, 5]
);
assert!(results.iter().all(|result| result.stable));
assert_eq!(write_calls.load(Ordering::SeqCst), 6);
drop(file);
let backend = client.into_inner().unwrap();
assert_eq!(&backend.data[..5], b"abcde");
assert_eq!(&backend.data[10..15], b"VWXYZ");
}
#[test]
fn write_allv_preserves_order_for_overlapping_short_writes() {
let client = FsClient::new(ScalarOnly {
max_write_once: Some(2),
..ScalarOnly::default()
});
let file = client
.open_with(OpenOp::new("/file", OpenFlags::WRITE))
.unwrap();
let results = client
.vwrite_all_native(&[
file.write_request_at(0, b"AAAA"),
file.write_request_at(2, b"BBBB"),
])
.unwrap();
assert_eq!(
results
.iter()
.map(|result| result.written)
.collect::<Vec<_>>(),
[4, 4]
);
drop(file);
assert_eq!(client.into_inner().unwrap().data, b"AABBBB");
}
#[test]
fn write_allv_waits_for_every_earlier_overlapping_request() {
let client = FsClient::new(ScalarOnly {
max_write_once: Some(2),
..ScalarOnly::default()
});
let file = client
.open_with(OpenOp::new("/file", OpenFlags::WRITE))
.unwrap();
client
.vwrite_all_native(&[
file.write_request_at(0, b"AAAAAA"),
file.write_request_at(6, b"BB"),
file.write_request_at(4, b"CCCC"),
])
.unwrap();
drop(file);
assert_eq!(client.into_inner().unwrap().data, b"AAAACCCC");
}
#[test]
fn write_allv_rejects_zero_progress() {
let client = FsClient::new(ScalarOnly {
max_write_once: Some(0),
..ScalarOnly::default()
});
let file = client
.open_with(OpenOp::new("/file", OpenFlags::WRITE))
.unwrap();
let error = client
.vwrite_all_native(&[file.write_request_at(0, b"data")])
.unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_IO);
assert_eq!(error.index(), Some(0));
}
#[test]
fn write_allv_validates_every_request_before_writing_a_prefix() {
let write_calls = Arc::new(AtomicUsize::new(0));
let client = FsClient::new(ScalarOnly {
write_calls: Arc::clone(&write_calls),
..ScalarOnly::default()
});
let other = FsClient::new(ScalarOnly::default());
let local_file = client
.open_with(OpenOp::new("/local", OpenFlags::WRITE))
.unwrap();
let foreign_file = other
.open_with(OpenOp::new("/foreign", OpenFlags::WRITE))
.unwrap();
let error = client
.vwrite_all_native(&[
local_file.write_request_at(0, b"would-write"),
foreign_file.write_request_at(0, b"invalid"),
])
.unwrap_err();
assert_eq!(error.index(), Some(1));
assert_eq!(write_calls.load(Ordering::SeqCst), 0);
}
#[test]
fn vector_results_must_match_their_requests_and_io_limits() {
for backend in [
ScalarOnly {
wrong_result_file: true,
..ScalarOnly::default()
},
ScalarOnly {
wrong_result_offset: true,
..ScalarOnly::default()
},
ScalarOnly {
oversized_read: true,
..ScalarOnly::default()
},
] {
let client = FsClient::new(backend);
let file = client.open("/file").unwrap();
let error = client
.vread_native(&[file.read_request_at(0, 1)])
.unwrap_err();
assert!(error.is_transport());
assert_eq!(error.index(), Some(0));
assert_eq!(error.operation(), Some("vread_native"));
drop(file);
}
let client = FsClient::new(ScalarOnly {
oversized_write_count: true,
..ScalarOnly::default()
});
let file = client
.open_with(OpenOp::new("/file", OpenFlags::WRITE))
.unwrap();
let error = client
.vwrite_native(&[file.write_request_at(0, b"x")])
.unwrap_err();
assert!(error.is_transport());
assert_eq!(error.index(), Some(0));
assert_eq!(error.operation(), Some("vwrite_native"));
}
#[test]
fn explicit_close_failure_keeps_handle_armed_for_drop_cleanup() {
let client = FsClient::new(ScalarOnly {
close_failures_remaining: 1,
..ScalarOnly::default()
});
let file = client.open("/file").unwrap();
assert!(file.close().is_err());
let backend = client.into_inner().expect("drop released the client");
assert_eq!(backend.close_calls, 2);
assert!(!backend.open);
}
#[test]
fn failed_try_close_retains_handle_for_explicit_retry() {
let client = FsClient::new(ScalarOnly {
close_failures_remaining: 1,
..ScalarOnly::default()
});
let mut file = client.open("/file").unwrap();
assert!(file.try_close().is_err());
assert!(file.try_close().is_ok());
assert!(file.try_close().is_ok());
drop(file);
let backend = client.into_inner().expect("drop released the client");
assert_eq!(backend.close_calls, 2);
assert!(!backend.open);
}
#[test]
fn failed_try_closev_retains_handles_for_explicit_retry() {
let client = FsClient::new(ScalarOnly {
close_failures_remaining: 1,
..ScalarOnly::default()
});
let mut files = vec![client.open("/file").unwrap()];
assert!(client.vclose(&mut files).is_err());
assert_eq!(files[0].path(), std::path::Path::new("/file"));
client.vclose(&mut files).unwrap();
drop(files);
let backend = client.into_inner().unwrap();
assert_eq!(backend.close_calls, 2);
assert!(!backend.open);
}
#[test]
fn successfully_closed_handle_returns_ebadf_instead_of_panicking() {
let client = FsClient::new(ScalarOnly::default());
let mut file = client.open("/file").unwrap();
file.try_close().unwrap();
let error = file.read_native(&mut [0u8; 1]).unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_EBADF);
let error = file.write_native(b"x").unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_EBADF);
let error = file.seek_native(SeekFrom::Start(0)).unwrap_err();
assert_eq!(error.err_no(), vfsi_sync::ERR_EBADF);
}
#[test]
fn stream_callback_can_reenter_client_and_drop_another_file() {
#[cfg(feature = "test-support")]
if vfsi_sync::test_support::supervise_with_deadline(
"stream_callback_can_reenter_client_and_drop_another_file",
) {
return;
}
let client = FsClient::new(ScalarOnly {
data: b"abcdef".to_vec(),
..ScalarOnly::default()
});
let mut other = Some(client.open("/other").unwrap());
let mut received = Vec::new();
client
.read_stream_with_options(
"/file",
vfsi_sync::StreamOptions::new().chunk_size(2),
|_, data| {
drop(other.take());
let file = client.open("/nested")?;
file.close()?;
received.extend_from_slice(data);
Ok(true)
},
)
.unwrap();
assert_eq!(received, b"abcdef");
}
#[test]
fn directory_visit_starts_with_one_entry_and_respects_tight_limits() {
let page_sizes = Arc::new(Mutex::new(Vec::new()));
let client = FsClient::new(ScalarOnly {
directory_entries: 2_100,
directory_page_sizes: Arc::clone(&page_sizes),
..ScalarOnly::default()
});
assert_eq!(
client
.visit_dir("/tree", |_| Ok(std::ops::ControlFlow::Break(())))
.unwrap(),
vfsi_sync::TraversalCompletion::Stopped
);
assert_eq!(*page_sizes.lock().unwrap(), [1]);
page_sizes.lock().unwrap().clear();
let mut seen = 0;
let error = client
.visit_dir_with_options(
"/tree",
vfsi_sync::ListDirOptions::new().max_entries(3),
|_| {
seen += 1;
Ok(std::ops::ControlFlow::Continue(()))
},
)
.unwrap_err();
assert_eq!(seen, 3);
assert_eq!(error.err_no(), libc::EFBIG as u32);
assert_eq!(*page_sizes.lock().unwrap(), [1, 3]);
page_sizes.lock().unwrap().clear();
let mut seen = 0;
let completion = client
.visit_dir("/tree", |_| {
seen += 1;
Ok(std::ops::ControlFlow::Continue(()))
})
.unwrap();
assert_eq!(completion, vfsi_sync::TraversalCompletion::Complete);
assert_eq!(seen, 2_100);
let page_sizes = page_sizes.lock().unwrap();
assert_eq!(page_sizes.first(), Some(&1));
assert_eq!(page_sizes.len(), 18);
assert!(page_sizes[1..].iter().all(|size| *size == 128));
let empty = FsClient::new(ScalarOnly::default());
assert_eq!(
empty
.visit_dir("/empty", |_| panic!("empty directory has no entry"))
.unwrap(),
vfsi_sync::TraversalCompletion::Complete
);
let single = FsClient::new(ScalarOnly {
directory_entries: 1,
..Default::default()
});
assert_eq!(
single
.visit_dir("/single", |_| Ok(std::ops::ControlFlow::Break(())))
.unwrap(),
vfsi_sync::TraversalCompletion::Stopped
);
}
#[test]
fn visitor_field_selection_is_forwarded_to_each_page() {
let fields = Arc::new(Mutex::new(Vec::new()));
let pages = Arc::new(Mutex::new(Vec::new()));
let client = FsClient::new(ScalarOnly {
directory_entries: 3,
directory_fields: fields.clone(),
directory_page_sizes: pages.clone(),
..Default::default()
});
let mut seen = 0;
client
.visit_dir_with_options(
"/tree",
vfsi_sync::ListDirOptions::new().fields(vfsi_sync::AttrMask::SIZE),
|_| {
seen += 1;
Ok(std::ops::ControlFlow::Continue(()))
},
)
.unwrap();
assert_eq!(seen, 3);
assert_eq!(
*fields.lock().unwrap(),
[vfsi_sync::AttrMask::MODE | vfsi_sync::AttrMask::SIZE; 2]
);
assert_eq!(*pages.lock().unwrap(), [1, 128]);
}
#[test]
fn native_read_into_dispatches_without_owned_read_results() {
let calls = Arc::new(AtomicUsize::new(0));
let client = FsClient::new(ScalarOnly {
data: b"abcdef".to_vec(),
direct_into_only: true,
into_calls: Arc::clone(&calls),
..ScalarOnly::default()
});
let file = client.open("/file").unwrap();
let mut first = [0u8; 3];
let mut second = [0u8; 3];
let lengths = client
.vread_into_native(&mut [
file.read_request_at_into(0, &mut first),
file.read_request_at_into(3, &mut second),
])
.unwrap();
assert_eq!(
lengths.iter().map(|result| result.read).collect::<Vec<_>>(),
[3, 3]
);
assert_eq!(
lengths
.iter()
.map(|result| result.offset)
.collect::<Vec<_>>(),
[0, 3]
);
assert!(!lengths[0].eof);
assert!(lengths[1].eof);
assert_eq!(&first, b"abc");
assert_eq!(&second, b"def");
let mut scalar = [0u8; 2];
assert_eq!(file.read_at(&mut scalar, 2).unwrap(), 2);
assert_eq!(&scalar, b"cd");
assert_eq!(calls.load(Ordering::SeqCst), 3);
}
#[test]
fn native_read_into_rejects_malformed_backend_results() {
let client = FsClient::new(ScalarOnly {
data: b"abc".to_vec(),
direct_into_only: true,
wrong_result_file: true,
..ScalarOnly::default()
});
let file = client.open("/file").unwrap();
let mut buffer = [0u8; 3];
let error = client
.vread_into_native(&mut [file.read_request_at_into(0, &mut buffer)])
.unwrap_err();
assert!(error.is_transport());
assert_eq!(error.index(), Some(0));
let backend = ScalarOnly {
data: b"abc".to_vec(),
direct_into_only: true,
..ScalarOnly::default()
};
let limit = Arc::clone(&backend.vector_result_limit);
let client = FsClient::new(backend);
let file = client.open("/file").unwrap();
*limit.lock().unwrap() = Some(0);
let error = client
.vread_into_native(&mut [file.read_request_at_into(0, &mut buffer)])
.unwrap_err();
assert!(error.is_transport());
assert_eq!(error.index(), None);
}
#[test]
fn notifications_reenter_and_drop_handles_after_unlock() {
let backend = ScalarOnly::default();
let notifications = backend.notifications.clone();
let observed = backend.close_observed.clone();
let client = FsClient::new(backend);
let file = client.open("/file").unwrap();
let callback_client = client.clone();
let (tx, rx) = std::sync::mpsc::channel();
notifications.lock().unwrap().push(Box::new(move || {
drop(file);
callback_client.capabilities().unwrap();
tx.send(()).unwrap();
}));
let worker = std::thread::spawn(move || {
client.capabilities().unwrap();
client
});
rx.recv_timeout(std::time::Duration::from_secs(2))
.expect("notification reentry must not deadlock");
let client = worker.join().unwrap();
client.drain_cleanup().unwrap();
assert_eq!(observed.load(Ordering::SeqCst), 1);
}
#[test]
fn file_drop_never_waits_for_an_in_flight_rpc() {
let entered = Arc::new(std::sync::Barrier::new(2));
let release = Arc::new(std::sync::Barrier::new(2));
let backend = ScalarOnly {
data: vec![1],
blocked_read: Some((entered.clone(), release.clone())),
..Default::default()
};
let closes = backend.close_observed.clone();
let client = FsClient::new(backend);
let mut reading = client.open("/file").unwrap();
let dropped = client.open("/other").unwrap();
let reader = std::thread::spawn(move || reading.read_native(&mut [0]).unwrap());
entered.wait();
let (tx, rx) = std::sync::mpsc::channel();
let dropper = std::thread::spawn(move || {
drop(dropped);
tx.send(()).unwrap();
});
let result = rx.recv_timeout(std::time::Duration::from_secs(1));
let before = closes.load(Ordering::SeqCst);
release.wait();
reader.join().unwrap();
dropper.join().unwrap();
result.expect("Drop must not wait for the backend mutex");
assert_eq!(before, 0, "Drop must not issue a remote close");
client.drain_cleanup().unwrap();
assert_eq!(closes.load(Ordering::SeqCst), 2);
}
#[test]
fn failed_deferred_close_keeps_ownership_until_drained() {
let backend = ScalarOnly {
close_failures_remaining: 1,
..Default::default()
};
let closes = backend.close_observed.clone();
let client = FsClient::new(backend);
drop(client.open("/file").unwrap());
assert_eq!(closes.load(Ordering::SeqCst), 0);
let error = client.drain_cleanup().unwrap_err();
assert_eq!(error.path(), Some(std::path::Path::new("/file")));
assert_eq!(error.operation(), Some("close"));
assert_eq!(closes.load(Ordering::SeqCst), 1);
client.drain_cleanup().unwrap();
client.drain_cleanup().unwrap();
assert_eq!(closes.load(Ordering::SeqCst), 2);
}
#[test]
fn observer_panic_does_not_poison_the_backend_mutex() {
let backend = ScalarOnly::default();
let notifications = backend.notifications.clone();
let client = FsClient::new(backend);
notifications
.lock()
.unwrap()
.push(Box::new(|| panic!("observer failure")));
assert!(client.capabilities().is_ok());
assert!(client.capabilities().is_ok());
}
#[test]
fn final_owner_drains_deferred_cleanup() {
let backend = ScalarOnly::default();
let closes = backend.close_observed.clone();
let client = FsClient::new(backend);
drop(client.open("/file").unwrap());
assert_eq!(closes.load(Ordering::SeqCst), 0);
drop(client);
assert_eq!(closes.load(Ordering::SeqCst), 1);
}
#[test]
fn ordinary_operation_drains_dropped_handles() {
let backend = ScalarOnly::default();
let closes = backend.close_observed.clone();
let client = FsClient::new(backend);
drop(client.open("/file").unwrap());
client.capabilities().unwrap();
assert_eq!(closes.load(Ordering::SeqCst), 1);
}
#[test]
fn filesystem_statistics_validate_backend_shape_and_preflight_all_handles() {
for count in [0, 1, 3] {
let client = FsClient::new(ScalarOnly {
stats_result_count: Some(count),
..Default::default()
});
assert!(client.vstatfs(&["/a", "/b"]).unwrap_err().is_transport());
}
let client = FsClient::new(ScalarOnly {
stats_error_index: Some(2),
..Default::default()
});
assert!(client.vstatfs(&["/a", "/b"]).unwrap_err().is_transport());
let backend = ScalarOnly::default();
let calls = backend.stats_calls.clone();
let client = FsClient::new(backend);
client.vstatfs::<&str>(&[]).unwrap();
assert_eq!(calls.load(Ordering::SeqCst), 0);
let mut file = client.open("/file").unwrap();
file.try_close().unwrap();
let targets = [
vfsi_sync::MetadataTarget::Path(std::path::Path::new("/a")),
vfsi_sync::MetadataTarget::File(&file),
];
assert_eq!(client.vstatfs(&targets).unwrap_err().index(), Some(1));
assert_eq!(calls.load(Ordering::SeqCst), 0);
let other = FsClient::new(ScalarOnly::default());
let foreign = other.open("/foreign").unwrap();
let targets = [
vfsi_sync::MetadataTarget::Path(std::path::Path::new("/a")),
vfsi_sync::MetadataTarget::File(&foreign),
];
assert_eq!(client.vstatfs(&targets).unwrap_err().index(), Some(1));
assert_eq!(calls.load(Ordering::SeqCst), 0);
}