use std::io;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use prov_graph::fs::ReadStorage;
pub mod memory;
pub use memory::InMemoryFs;
pub use prov_graph::fs::StdFs;
pub trait Storage: ReadStorage {
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 capabilities(&self) -> Capabilities {
Capabilities::NONE
}
fn sync(&self, path: &Path, need: Durability) -> impl Future<Output = io::Result<()>> {
async move {
let _ = (path, need);
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?;
self.sync(path, Durability::Durable).await?;
return match parent_dir(path) {
Some(dir) => self.sync(dir, Durability::Durable).await,
None => Ok(()),
};
}
let tmp = temp_sibling(path);
let staged = async {
self.write(&tmp, contents).await?;
self.sync(&tmp, Durability::Ordered).await?;
self.rename(&tmp, path).await
}
.await;
match staged {
Ok(()) => match parent_dir(path) {
Some(dir) => self.sync(dir, Durability::Durable).await,
None => Ok(()),
},
Err(e) => {
let _ = self.remove_file(&tmp).await;
Err(e)
}
}
}
}
}
impl<S: Storage + ?Sized> Storage for &S {
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
(**self).write(path, contents).await
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
(**self).create_dir_all(path).await
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
(**self).remove_file(path).await
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
(**self).remove_dir_all(path).await
}
async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
(**self).rename(from, to).await
}
fn capabilities(&self) -> Capabilities {
(**self).capabilities()
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
(**self).sync(path, need).await
}
async fn write_atomic(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
(**self).write_atomic(path, contents).await
}
}
impl<S: Storage + ?Sized> Storage for Arc<S> {
async fn write(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
(**self).write(path, contents).await
}
async fn create_dir_all(&self, path: &Path) -> io::Result<()> {
(**self).create_dir_all(path).await
}
async fn remove_file(&self, path: &Path) -> io::Result<()> {
(**self).remove_file(path).await
}
async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
(**self).remove_dir_all(path).await
}
async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
(**self).rename(from, to).await
}
fn capabilities(&self) -> Capabilities {
(**self).capabilities()
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
(**self).sync(path, need).await
}
async fn write_atomic(&self, path: &Path, contents: &[u8]) -> io::Result<()> {
(**self).write_atomic(path, contents).await
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Capabilities {
pub atomic_replace: bool,
pub sync_guarantee: SyncGuarantee,
pub native_transactions: bool,
}
impl Capabilities {
pub const NONE: Self = Self {
atomic_replace: false,
sync_guarantee: SyncGuarantee::None,
native_transactions: false,
};
pub const LOCAL_FS: Self = Self {
atomic_replace: true,
sync_guarantee: SyncGuarantee::Durable,
native_transactions: false,
};
pub const IN_MEMORY: Self = Self {
atomic_replace: true,
sync_guarantee: SyncGuarantee::None,
native_transactions: false,
};
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Durability {
Ordered,
Durable,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum SyncGuarantee {
None,
Ordered,
Durable,
}
impl SyncGuarantee {
pub const fn satisfies(self, need: Durability) -> bool {
match need {
Durability::Ordered => !matches!(self, SyncGuarantee::None),
Durability::Durable => matches!(self, SyncGuarantee::Durable),
}
}
}
fn parent_dir(path: &Path) -> Option<&Path> {
path.parent().filter(|p| !p.as_os_str().is_empty())
}
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"))
}
impl Storage for StdFs {
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)
}
fn capabilities(&self) -> Capabilities {
Capabilities::LOCAL_FS
}
async fn sync(&self, path: &Path, need: Durability) -> io::Result<()> {
let _ = need;
sync_path(path)
}
}
fn sync_path(path: &Path) -> io::Result<()> {
#[cfg(not(unix))]
if path.is_dir() {
return Ok(());
}
match std::fs::File::open(path) {
Ok(file) => file.sync_all()?,
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => return Err(e),
}
Ok(())
}
#[cfg(test)]
mod tests {
use prov_graph::exec::block_on;
use super::*;
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_eq!(StdFs.capabilities().sync_guarantee, SyncGuarantee::Durable);
assert!(!StdFs.capabilities().native_transactions);
}
#[test]
fn a_guarantee_answers_only_the_requests_it_can_keep() {
assert!(!SyncGuarantee::None.satisfies(Durability::Ordered));
assert!(!SyncGuarantee::None.satisfies(Durability::Durable));
assert!(SyncGuarantee::Ordered.satisfies(Durability::Ordered));
assert!(!SyncGuarantee::Ordered.satisfies(Durability::Durable));
assert!(SyncGuarantee::Durable.satisfies(Durability::Ordered));
assert!(SyncGuarantee::Durable.satisfies(Durability::Durable));
}
#[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"), Durability::Durable)).unwrap();
}
#[test]
fn sync_flushes_a_directory_as_readily_as_a_file() {
let root = tmp("sync-dir");
block_on(StdFs.sync(&root, Durability::Durable)).unwrap();
}
}