sausage/vfs/src/journal/layer.rs

//! Tagged mutations and writable descriptors. Read-only calls directly reach the inner layer.
use super::*;
use crate::{Entry as DirEntry, File, Process, Stat, async_trait};
pub(super) struct View {
    pub journal: Arc<Journal>,
    pub tag: Tag,
}
impl View {
    async fn mutate<T: Send + 'static>(
        &self,
        paths: Vec<String>,
        operation: impl Future<Output = Result<T>> + Send + 'static,
    ) -> Result<T> {
        let (journal, tag) = (self.journal.clone(), self.tag.clone());
        Process::default()
            .operation(async move {
                let _gate = journal.gate.lock().await;
                for path in paths {
                    journal.capture(&tag, &path).await?;
                }
                operation.await
            })
            .await
    }
}
#[async_trait]
impl Layer for View {
    fn observer(&self) -> Option<crate::observe::Observer> {
        self.journal.root.observer()
    }
    fn journal(&self) -> Option<Arc<Journal>> {
        Some(self.journal.clone())
    }
    fn tagged(&self, tag: Tag) -> Option<Arc<dyn Layer>> {
        Some(self.journal.layer(tag))
    }
    async fn stat(&self, p: &str, f: bool) -> Result<Stat> {
        self.journal.root.stat(p, f).await
    }
    async fn readlink(&self, p: &str) -> Result<String> {
        self.journal.root.readlink(p).await
    }
    async fn list(&self, p: &str) -> Result<Vec<DirEntry>> {
        self.journal.root.list(p).await
    }
    async fn open(&self, p: &str, o: Open) -> Result<Arc<dyn File>> {
        if !(o.write || o.create || o.truncate) || Journal::excluded(p) {
            return self.journal.root.open(p, o).await;
        }
        let (journal, tag, path) = (self.journal.clone(), self.tag.clone(), p.to_owned());
        Process::default()
            .operation(async move {
                let _gate = journal.gate.lock().await;
                if o.create || o.truncate {
                    journal.capture(&tag, &path).await?;
                }
                let file = journal.root.open(&path, o).await?;
                let path = Arc::new(Mutex::new(Some(path)));
                let mut paths = journal.paths.lock().unwrap();
                paths.retain(|p| p.strong_count() != 0);
                paths.push(Arc::downgrade(&path));
                Ok(Arc::new(Handle {
                    file,
                    path,
                    view: View {
                        journal: journal.clone(),
                        tag,
                    },
                }) as Arc<dyn File>)
            })
            .await
    }
    async fn mkdir(&self, p: &str) -> Result<()> {
        let (root, path) = (self.journal.root.clone(), p.to_owned());
        self.mutate(vec![path.clone()], async move { root.mkdir(&path).await })
            .await
    }
    async fn remove_file(&self, p: &str) -> Result<()> {
        self.remove(p, false).await
    }
    async fn remove_dir(&self, p: &str) -> Result<()> {
        self.remove(p, true).await
    }
    async fn rename(&self, a: &str, b: &str) -> Result<()> {
        let (journal, tag, a, b) = (
            self.journal.clone(),
            self.tag.clone(),
            a.to_owned(),
            b.to_owned(),
        );
        Process::default()
            .operation(async move {
                let _gate = journal.gate.lock().await;
                let mut pending = vec![(a.clone(), b.clone())];
                while let Some((from, to)) = pending.pop() {
                    journal.capture(&tag, &from).await?;
                    journal.capture(&tag, &to).await?;
                    if journal.root.stat(&from, false).await?.directory {
                        for entry in journal.root.list(&from).await? {
                            pending.push((
                                format!("{from}/{}", entry.name),
                                format!("{to}/{}", entry.name),
                            ));
                        }
                    }
                }
                journal.root.rename(&a, &b).await?;
                if a != b {
                    journal.moved(&a, Some(&b));
                }
                Ok(())
            })
            .await
    }
}
impl View {
    async fn remove(&self, p: &str, directory: bool) -> Result<()> {
        let (journal, path) = (self.journal.clone(), p.to_owned());
        self.mutate(vec![path.clone()], async move {
            if directory {
                journal.root.remove_dir(&path).await?;
            } else {
                journal.root.remove_file(&path).await?;
            }
            journal.moved(&path, None);
            Ok(())
        })
        .await
    }
}
struct Handle {
    file: Arc<dyn File>,
    path: Arc<Mutex<Option<String>>>,
    view: View,
}
impl Handle {
    async fn mutate<T: Send + 'static>(
        &self,
        operation: impl Future<Output = Result<T>> + Send + 'static,
    ) -> Result<T> {
        let (journal, tag, path) = (
            self.view.journal.clone(),
            self.view.tag.clone(),
            self.path.clone(),
        );
        Process::default()
            .operation(async move {
                let _gate = journal.gate.lock().await;
                let path = path.lock().unwrap().clone();
                if let Some(path) = path {
                    journal.capture(&tag, &path).await?;
                }
                operation.await
            })
            .await
    }
}
#[async_trait]
impl File for Handle {
    async fn stat(&self) -> Result<Stat> {
        self.file.stat().await
    }
    async fn read(&self, o: u64, n: usize) -> Result<Vec<u8>> {
        self.file.read(o, n).await
    }
    async fn write(&self, o: u64, b: Vec<u8>) -> Result<usize> {
        let f = self.file.clone();
        self.mutate(async move { f.write(o, b).await }).await
    }
    async fn append(&self, b: Vec<u8>) -> Result<usize> {
        let f = self.file.clone();
        self.mutate(async move { f.append(b).await }).await
    }
    async fn set_size(&self, n: u64) -> Result<()> {
        let f = self.file.clone();
        self.mutate(async move { f.set_size(n).await }).await
    }
    async fn sync(&self) -> Result<()> {
        self.file.sync().await
    }
    async fn set_times(&self, a: Option<SystemTime>, m: Option<SystemTime>) -> Result<()> {
        let f = self.file.clone();
        self.mutate(async move { f.set_times(a, m).await }).await
    }
}
impl Journal {
    pub(super) fn moved(&self, from: &str, to: Option<&str>) {
        let mut paths = self.paths.lock().unwrap();
        paths.retain(|p| {
            let Some(p) = p.upgrade() else { return false };
            let mut path = p.lock().unwrap();
            if let Some(old) = path.as_deref() {
                if under(old, from) {
                    *path = to.map(|to| format!("{to}{}", &old[from.len()..]));
                } else if to.is_some_and(|to| under(old, to)) {
                    *path = None;
                }
            }
            true
        });
    }
}