#![cfg(test)]
use std::io;
use std::path::{Path, PathBuf};
use prov_graph::fs::{DirEntry, Metadata, ReadStorage, StdFs};
use prov_store::fs::{Capabilities, Durability, Storage, SyncGuarantee};
#[derive(Debug)]
pub(crate) struct FailAtWrite {
writes: std::cell::Cell<usize>,
fail_at: usize,
}
impl FailAtWrite {
pub(crate) fn nth(fail_at: usize) -> Self {
Self {
writes: std::cell::Cell::new(0),
fail_at,
}
}
pub(crate) fn never() -> Self {
Self::nth(usize::MAX)
}
pub(crate) fn attempted(&self) -> usize {
self.writes.get()
}
}
macro_rules! reads_like_stdfs {
($ty:ty) => {
impl ReadStorage for $ty {
async fn read(&self, path: &Path) -> io::Result<Vec<u8>> {
StdFs.read(path).await
}
async fn read_to_string(&self, path: &Path) -> io::Result<String> {
StdFs.read_to_string(path).await
}
async fn read_dir(&self, path: &Path) -> io::Result<Vec<DirEntry>> {
StdFs.read_dir(path).await
}
async fn metadata(&self, path: &Path) -> io::Result<Metadata> {
StdFs.metadata(path).await
}
}
};
}
reads_like_stdfs!(FailAtWrite);
impl Storage for FailAtWrite {
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
if crate::journal::is_journal_path(path) {
return StdFs.write(path, contents).await;
}
let n = self.writes.get();
self.writes.set(n + 1);
if n == self.fail_at {
return Err(io::Error::other("disk full (test)"));
}
StdFs.write(path, contents).await
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.create_dir_all(path).await
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
StdFs.remove_file(path).await
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.remove_dir_all(path).await
}
async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
StdFs.rename(from, to).await
}
fn capabilities(&self) -> Capabilities {
Capabilities::LOCAL_FS
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
StdFs.sync(path, need).await
}
}
#[derive(Debug)]
pub(crate) struct RecordingFs {
log: std::cell::RefCell<Vec<FsEvent>>,
caps: Capabilities,
}
#[derive(Clone, Default, Debug)]
pub(crate) struct CountingFs {
docs: std::sync::Arc<std::sync::Mutex<std::collections::BTreeMap<PathBuf, usize>>>,
}
impl CountingFs {
pub(crate) fn doc_reads(&self, dir: &Path, rel: &str) -> usize {
count(&self.docs, &dir.join(rel))
}
}
fn count(
counter: &std::sync::Mutex<std::collections::BTreeMap<PathBuf, usize>>,
path: &Path,
) -> usize {
counter.lock().unwrap().get(path).copied().unwrap_or(0)
}
fn tally(counter: &std::sync::Mutex<std::collections::BTreeMap<PathBuf, usize>>, path: &Path) {
*counter
.lock()
.unwrap()
.entry(path.to_path_buf())
.or_insert(0) += 1;
}
impl ReadStorage for CountingFs {
async fn read(&self, path: &Path) -> io::Result<Vec<u8>> {
StdFs.read(path).await
}
async fn read_to_string(&self, path: &Path) -> io::Result<String> {
tally(&self.docs, path);
StdFs.read_to_string(path).await
}
async fn read_dir(&self, path: &Path) -> io::Result<Vec<DirEntry>> {
StdFs.read_dir(path).await
}
async fn metadata(&self, path: &Path) -> io::Result<Metadata> {
StdFs.metadata(path).await
}
}
impl Storage for CountingFs {
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
StdFs.write(path, contents).await
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.create_dir_all(path).await
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
StdFs.remove_file(path).await
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.remove_dir_all(path).await
}
async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
StdFs.rename(from, to).await
}
fn capabilities(&self) -> Capabilities {
StdFs.capabilities()
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
StdFs.sync(path, need).await
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum FsEvent {
Write(PathBuf),
Sync(PathBuf, Durability),
Rename(PathBuf, PathBuf),
Remove(PathBuf),
}
impl RecordingFs {
pub(crate) fn local() -> Self {
Self {
log: std::cell::RefCell::new(Vec::new()),
caps: Capabilities::LOCAL_FS,
}
}
pub(crate) fn with_caps(caps: Capabilities) -> Self {
Self {
log: std::cell::RefCell::new(Vec::new()),
caps,
}
}
pub(crate) fn events(&self) -> Vec<FsEvent> {
self.log.borrow().clone()
}
}
reads_like_stdfs!(RecordingFs);
impl Storage for RecordingFs {
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
self.log
.borrow_mut()
.push(FsEvent::Write(path.to_path_buf()));
StdFs.write(path, contents).await
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.create_dir_all(path).await
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
self.log
.borrow_mut()
.push(FsEvent::Remove(path.to_path_buf()));
StdFs.remove_file(path).await
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.remove_dir_all(path).await
}
async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
self.log
.borrow_mut()
.push(FsEvent::Rename(from.to_path_buf(), to.to_path_buf()));
StdFs.rename(from, to).await
}
fn capabilities(&self) -> Capabilities {
self.caps
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
self.log
.borrow_mut()
.push(FsEvent::Sync(path.to_path_buf(), need));
StdFs.sync(path, need).await
}
}
#[derive(Debug)]
pub(crate) struct FailingRename;
reads_like_stdfs!(FailingRename);
impl Storage for FailingRename {
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
StdFs.write(path, contents).await
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.create_dir_all(path).await
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
StdFs.remove_file(path).await
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
StdFs.remove_dir_all(path).await
}
async fn rename(&self, _from: &Path, _to: &Path) -> io::Result<()> {
Err(io::Error::other("rename failed (test)"))
}
fn capabilities(&self) -> Capabilities {
Capabilities::LOCAL_FS
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
StdFs.sync(path, need).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use prov_graph::exec::block_on;
fn tmp(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("prov-fsfault-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
#[test]
fn a_torn_atomic_write_leaves_the_target_untouched_and_no_litter() {
let root = tmp("atomic-fail");
std::fs::write(root.join("doc.md"), "old").unwrap();
let target = root.join("doc.md");
let temp = root.join(".doc.md.prov-tmp");
let err = block_on(FailingRename.write_atomic(&target, b"new")).unwrap_err();
assert!(err.to_string().contains("rename failed"), "{err}");
assert_eq!(
std::fs::read_to_string(&target).unwrap(),
"old",
"target was touched"
);
assert!(!temp.exists(), "the staging file was left behind");
}
#[test]
fn the_default_capability_promises_nothing() {
let bare = RecordingFs::with_caps(Capabilities::NONE);
assert_eq!(bare.capabilities(), Capabilities::NONE);
assert_eq!(
Capabilities::NONE,
Capabilities {
atomic_replace: false,
sync_guarantee: SyncGuarantee::None,
native_transactions: false
}
);
}
#[test]
fn write_atomic_follows_the_durable_replace_protocol() {
let root = tmp("protocol");
std::fs::write(root.join("doc.md"), "old").unwrap();
let fs = RecordingFs::local();
let target = root.join("doc.md");
let temp = root.join(".doc.md.prov-tmp");
block_on(fs.write_atomic(&target, b"new")).unwrap();
assert_eq!(std::fs::read_to_string(&target).unwrap(), "new");
assert!(!temp.exists(), "the staging file must be gone");
assert_eq!(
fs.events(),
vec![
FsEvent::Write(temp.clone()),
FsEvent::Sync(temp.clone(), Durability::Ordered),
FsEvent::Rename(temp.clone(), target.clone()),
FsEvent::Sync(root.clone(), Durability::Durable),
],
"the atomic-replace protocol ran out of order"
);
}
#[test]
fn write_atomic_creates_a_new_file_without_disturbing_the_directory() {
let root = tmp("atomic-create");
let fs = RecordingFs::local();
let target = root.join("fresh.md");
let temp = root.join(".fresh.md.prov-tmp");
block_on(fs.write_atomic(&target, b"hello")).unwrap();
assert_eq!(std::fs::read_to_string(&target).unwrap(), "hello");
assert!(!temp.exists());
assert_eq!(fs.events().first(), Some(&FsEvent::Write(temp)));
}
#[test]
fn a_non_atomic_backend_writes_straight_through_without_claiming_atomicity() {
let root = tmp("fallback");
let fs = RecordingFs::with_caps(Capabilities::NONE);
let target = root.join("doc.md");
block_on(fs.write_atomic(&target, b"hello")).unwrap();
assert_eq!(std::fs::read_to_string(&target).unwrap(), "hello");
assert_eq!(
fs.events(),
vec![
FsEvent::Write(target.clone()),
FsEvent::Sync(target, Durability::Durable),
FsEvent::Sync(root.clone(), Durability::Durable),
],
"the fallback must write the target directly, with no staging rename"
);
assert!(!root.join(".doc.md.prov-tmp").exists());
}
}