use std::io;
use std::path::{Path, PathBuf};
use std::time::SystemTime;
pub trait Storage {
fn read(&self, path: &Path) -> impl Future<Output = io::Result<Vec<u8>>>;
fn read_to_string(&self, path: &Path) -> impl Future<Output = io::Result<String>>;
fn read_dir(&self, path: &Path) -> impl Future<Output = io::Result<Vec<DirEntry>>>;
fn write(&self, path: &Path, contents: &[u8]) -> impl Future<Output = io::Result<()>>;
fn create_dir_all(&self, path: &Path) -> impl Future<Output = io::Result<()>>;
fn remove_file(&self, path: &Path) -> impl Future<Output = io::Result<()>>;
fn remove_dir_all(&self, path: &Path) -> impl Future<Output = io::Result<()>>;
fn rename(&self, from: &Path, to: &Path) -> impl Future<Output = io::Result<()>>;
fn metadata(&self, path: &Path) -> impl Future<Output = io::Result<Metadata>>;
fn try_exists(&self, path: &Path) -> impl Future<Output = io::Result<bool>> {
async move {
match self.metadata(path).await {
Ok(_) => Ok(true),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(false),
Err(e) => Err(e),
}
}
}
fn capabilities(&self) -> Capabilities {
Capabilities::NONE
}
fn sync(&self, path: &Path) -> impl Future<Output = io::Result<()>> {
async move {
let _ = path;
Ok(())
}
}
fn write_atomic(&self, path: &Path, contents: &[u8]) -> impl Future<Output = io::Result<()>> {
async move {
if !self.capabilities().atomic_replace {
self.write(path, contents).await?;
return self.sync(path).await;
}
let tmp = temp_sibling(path);
let staged = async {
self.write(&tmp, contents).await?;
self.sync(&tmp).await?;
self.rename(&tmp, path).await
}
.await;
match staged {
Ok(()) => self.sync(path).await,
Err(e) => {
let _ = self.remove_file(&tmp).await;
Err(e)
}
}
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Capabilities {
pub atomic_replace: bool,
pub durable_sync: bool,
pub native_transactions: bool,
}
impl Capabilities {
pub const NONE: Self = Self {
atomic_replace: false,
durable_sync: false,
native_transactions: false,
};
pub const LOCAL_FS: Self = Self {
atomic_replace: true,
durable_sync: true,
native_transactions: false,
};
}
fn temp_sibling(path: &Path) -> PathBuf {
let name = path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("document");
path.with_file_name(format!(".{name}.prov-tmp"))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DirEntry {
path: PathBuf,
file_type: FileType,
}
impl DirEntry {
pub fn new(path: impl Into<PathBuf>, file_type: FileType) -> Self {
Self {
path: path.into(),
file_type,
}
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn file_name(&self) -> Option<&std::ffi::OsStr> {
self.path.file_name()
}
pub fn file_type(&self) -> FileType {
self.file_type
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Metadata {
file_type: FileType,
len: u64,
modified: Option<SystemTime>,
}
impl Metadata {
pub fn new(file_type: FileType, len: u64, modified: Option<SystemTime>) -> Self {
Self {
file_type,
len,
modified,
}
}
pub fn file_type(&self) -> FileType {
self.file_type
}
pub fn is_file(&self) -> bool {
self.file_type.is_file()
}
pub fn is_dir(&self) -> bool {
self.file_type.is_dir()
}
pub fn len(&self) -> u64 {
self.len
}
pub fn is_empty(&self) -> bool {
self.len == 0
}
pub fn modified(&self) -> io::Result<SystemTime> {
self.modified
.ok_or_else(|| io::Error::new(io::ErrorKind::Unsupported, "modified time unavailable"))
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct StdFs;
impl Storage for StdFs {
async fn read(&self, path: &Path) -> io::Result<Vec<u8>> {
std::fs::read(path)
}
async fn read_to_string(&self, path: &Path) -> io::Result<String> {
std::fs::read_to_string(path)
}
async fn read_dir(&self, path: &Path) -> io::Result<Vec<DirEntry>> {
std::fs::read_dir(path)?
.map(|entry| {
let entry = entry?;
Ok(DirEntry::new(
entry.path(),
convert_file_type(entry.file_type()?),
))
})
.collect()
}
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
std::fs::write(path, contents)
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
std::fs::create_dir_all(path)
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
std::fs::remove_file(path)
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
std::fs::remove_dir_all(path)
}
async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
std::fs::rename(from, to)
}
async fn metadata(&self, path: &Path) -> io::Result<Metadata> {
let md = std::fs::metadata(path)?;
Ok(Metadata::new(
convert_file_type(md.file_type()),
md.len(),
md.modified().ok(),
))
}
fn capabilities(&self) -> Capabilities {
Capabilities::LOCAL_FS
}
async fn sync(&self, path: &Path) -> io::Result<()> {
sync_path(path)
}
}
fn sync_path(path: &Path) -> io::Result<()> {
match std::fs::File::open(path) {
Ok(file) => file.sync_all()?,
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => return Err(e),
}
#[cfg(unix)]
if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
match std::fs::File::open(parent) {
Ok(dir) => dir.sync_all()?,
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => return Err(e),
}
}
Ok(())
}
fn convert_file_type(ft: std::fs::FileType) -> FileType {
if ft.is_dir() {
FileType::DIR
} else if ft.is_file() {
FileType::FILE
} else {
FileType::SYMLINK
}
}
#[cfg(test)]
#[derive(Debug)]
pub(crate) struct FailAtWrite {
writes: std::cell::Cell<usize>,
fail_at: usize,
}
#[cfg(test)]
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()
}
}
#[cfg(test)]
impl Storage for FailAtWrite {
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 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
}
async fn metadata(&self, path: &Path) -> io::Result<Metadata> {
StdFs.metadata(path).await
}
fn capabilities(&self) -> Capabilities {
Capabilities::LOCAL_FS
}
async fn sync(&self, path: &Path) -> io::Result<()> {
StdFs.sync(path).await
}
}
#[cfg(test)]
#[derive(Debug)]
pub(crate) struct RecordingFs {
log: std::cell::RefCell<Vec<FsEvent>>,
caps: Capabilities,
}
#[cfg(test)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum FsEvent {
Write(PathBuf),
Sync(PathBuf),
Rename(PathBuf, PathBuf),
Remove(PathBuf),
}
#[cfg(test)]
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()
}
}
#[cfg(test)]
impl Storage for RecordingFs {
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 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
}
async fn metadata(&self, path: &Path) -> io::Result<Metadata> {
StdFs.metadata(path).await
}
fn capabilities(&self) -> Capabilities {
self.caps
}
async fn sync(&self, path: &Path) -> io::Result<()> {
self.log
.borrow_mut()
.push(FsEvent::Sync(path.to_path_buf()));
StdFs.sync(path).await
}
}
#[cfg(test)]
#[derive(Debug)]
pub(crate) struct FailingRename;
#[cfg(test)]
impl Storage for FailingRename {
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 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)"))
}
async fn metadata(&self, path: &Path) -> io::Result<Metadata> {
StdFs.metadata(path).await
}
fn capabilities(&self) -> Capabilities {
Capabilities::LOCAL_FS
}
async fn sync(&self, path: &Path) -> io::Result<()> {
StdFs.sync(path).await
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FileType {
is_dir: bool,
is_file: bool,
is_symlink: bool,
}
impl FileType {
pub const FILE: FileType = FileType {
is_dir: false,
is_file: true,
is_symlink: false,
};
pub const DIR: FileType = FileType {
is_dir: true,
is_file: false,
is_symlink: false,
};
pub const SYMLINK: FileType = FileType {
is_dir: false,
is_file: false,
is_symlink: true,
};
pub fn is_file(&self) -> bool {
self.is_file
}
pub fn is_dir(&self) -> bool {
self.is_dir
}
pub fn is_symlink(&self) -> bool {
self.is_symlink
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::exec::block_on;
fn tmp(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("prov-fs-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
#[test]
fn stdfs_declares_the_local_filesystem_guarantees() {
assert_eq!(StdFs.capabilities(), Capabilities::LOCAL_FS);
assert!(StdFs.capabilities().atomic_replace);
assert!(StdFs.capabilities().durable_sync);
assert!(!StdFs.capabilities().native_transactions);
}
#[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,
durable_sync: false,
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()),
FsEvent::Rename(temp.clone(), target.clone()),
FsEvent::Sync(target.clone()),
],
"the atomic-replace protocol ran out of order"
);
}
#[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 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)],
"the fallback must write the target directly, with no staging rename"
);
assert!(!root.join(".doc.md.prov-tmp").exists());
}
#[test]
fn sync_of_a_missing_path_is_not_an_error() {
let root = tmp("sync-missing");
block_on(StdFs.sync(&root.join("never-created.md"))).unwrap();
}
}