From 984c0486e0dcfced661297b87dc6ed8b080a33d3 Mon Sep 17 00:00:00 2001 From: simonsan <14062932+simonsan@users.noreply.github.com> Date: Sun, 24 Nov 2024 10:03:21 +0100 Subject: [PATCH] Revert "fix(filesystem): Use channels to communicate within webdav filesystem (#1361)" This reverts commit d79e392b86c79d73a91b7fe15b10a535d1110e9c. --- src/commands/webdav.rs | 6 +- src/commands/webdav/webdavfs.rs | 306 ++++++++++---------------------- 2 files changed, 93 insertions(+), 219 deletions(-) diff --git a/src/commands/webdav.rs b/src/commands/webdav.rs index aa06d39..90b7907 100644 --- a/src/commands/webdav.rs +++ b/src/commands/webdav.rs @@ -3,9 +3,6 @@ // ignore markdown clippy lints as we use doc-comments to generate clap help texts #![allow(clippy::doc_markdown)] -mod webdavfs; -use webdavfs::WebDavFS; - use std::net::ToSocketAddrs; use crate::{repository::CliIndexedRepo, status_err, Application, RusticConfig, RUSTIC_APP}; @@ -16,6 +13,9 @@ use dav_server::{warp::dav_handler, DavHandler}; use serde::{Deserialize, Serialize}; use rustic_core::vfs::{FilePolicy, IdenticalSnapshot, Latest, Vfs}; +use webdavfs::WebDavFS; + +mod webdavfs; #[derive(Clone, Command, Default, Debug, clap::Parser, Serialize, Deserialize, Merge)] #[serde(default, rename_all = "kebab-case", deny_unknown_fields)] diff --git a/src/commands/webdav/webdavfs.rs b/src/commands/webdav/webdavfs.rs index 87146ad..0c3ebb5 100644 --- a/src/commands/webdav/webdavfs.rs +++ b/src/commands/webdav/webdavfs.rs @@ -3,7 +3,7 @@ use std::os::unix::ffi::OsStrExt; use std::{ fmt::{Debug, Formatter}, io::SeekFrom, - sync::OnceLock, + sync::{Arc, OnceLock}, time::SystemTime, }; @@ -16,7 +16,6 @@ use dav_server::{ }, }; use futures::FutureExt; -use tokio::sync::{mpsc, oneshot}; use rustic_core::{ repofile::Node, @@ -41,7 +40,45 @@ struct DavFsInner { file_policy: FilePolicy, } -impl DavFsInner { +impl Debug for DavFsInner { + fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> { + write!(f, "DavFS") + } +} + +/// DAV Filesystem implementation. +/// +/// This is the main entry point for the DAV filesystem. +/// It implements [`DavFileSystem`] and can be used to serve a [`Repository`] via DAV. +#[derive(Debug)] +pub struct WebDavFS { + inner: Arc>, +} + +impl WebDavFS { + /// Create a new [`WebDavFS`] instance. + /// + /// # Arguments + /// + /// * `repo` - The [`Repository`] to use + /// * `vfs` - The [`Vfs`] to use + /// * `file_policy` - The [`FilePolicy`] to use + /// + /// # Returns + /// + /// A new [`WebDavFS`] instance + pub(crate) fn new(repo: Repository, vfs: Vfs, file_policy: FilePolicy) -> Self { + let inner = DavFsInner { + repo, + vfs, + file_policy, + }; + + Self { + inner: Arc::new(inner), + } + } + /// Get a [`Node`] from the specified [`DavPath`]. /// /// # Arguments @@ -58,8 +95,9 @@ impl DavFsInner { /// /// [`Tree`]: crate::repofile::Tree fn node_from_path(&self, path: &DavPath) -> Result { - self.vfs - .node_from_path(&self.repo, &path.as_pathbuf()) + self.inner + .vfs + .node_from_path(&self.inner.repo, &path.as_pathbuf()) .map_err(|_| FsError::GeneralFailure) } @@ -79,210 +117,31 @@ impl DavFsInner { /// /// [`Tree`]: crate::repofile::Tree fn dir_entries_from_path(&self, path: &DavPath) -> Result, FsError> { - self.vfs - .dir_entries_from_path(&self.repo, &path.as_pathbuf()) + self.inner + .vfs + .dir_entries_from_path(&self.inner.repo, &path.as_pathbuf()) .map_err(|_| FsError::GeneralFailure) } +} - fn open(&self, node: &Node, options: OpenOptions) -> Result { - if options.write - || options.append - || options.truncate - || options.create - || options.create_new - { - return Err(FsError::Forbidden); +impl Clone for WebDavFS { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), } - - if matches!(self.file_policy, FilePolicy::Forbidden) { - return Err(FsError::Forbidden); - } - - let open = self - .repo - .open_file(node) - .map_err(|_err| FsError::GeneralFailure)?; - Ok(open) - } - - fn read_bytes( - &self, - file: OpenFile, - seek: usize, - count: usize, - ) -> Result<(Bytes, OpenFile), FsError> { - let data = self - .repo - .read_file_at(&file, seek, count) - .map_err(|_err| FsError::GeneralFailure)?; - Ok((data, file)) } } -/// Messages used -#[allow(clippy::large_enum_variant)] -enum DavFsInnerCommand { - Node(DavPath, oneshot::Sender>), - DirEntries(DavPath, oneshot::Sender, FsError>>), - Open( - Node, - OpenOptions, - oneshot::Sender>, - ), - ReadBytes( - OpenFile, - usize, - usize, - oneshot::Sender>, - ), -} - -impl Debug for DavFsInner { - fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> { - write!(f, "DavFS") - } -} - -/// DAV Filesystem implementation. -/// -/// This is the main entry point for the DAV filesystem. -/// It implements [`DavFileSystem`] and can be used to serve a [`Repository`] via DAV. -#[derive(Debug, Clone)] -pub struct WebDavFS { - send: mpsc::Sender, -} - -impl WebDavFS { - /// Create a new [`WebDavFS`] instance. - /// - /// # Arguments - /// - /// * `repo` - The [`Repository`] to use - /// * `vfs` - The [`Vfs`] to use - /// * `file_policy` - The [`FilePolicy`] to use - /// - /// # Returns - /// - /// A new [`WebDavFS`] instance - pub(crate) fn new( - repo: Repository, - vfs: Vfs, - file_policy: FilePolicy, - ) -> Self { - let inner = DavFsInner { - repo, - vfs, - file_policy, - }; - - let (send, mut rcv) = mpsc::channel(1); - - let _ = std::thread::spawn(move || -> Result<_, FsError> { - while let Some(task) = rcv.blocking_recv() { - match task { - DavFsInnerCommand::Node(path, res) => { - res.send(inner.node_from_path(&path)) - .map_err(|_err| FsError::GeneralFailure)?; - } - DavFsInnerCommand::DirEntries(path, res) => { - res.send(inner.dir_entries_from_path(&path)) - .map_err(|_err| FsError::GeneralFailure)?; - } - DavFsInnerCommand::Open(path, open_options, res) => { - res.send(inner.open(&path, open_options)) - .map_err(|_err| FsError::GeneralFailure)?; - } - DavFsInnerCommand::ReadBytes(file, seek, count, res) => { - res.send(inner.read_bytes(file, seek, count)) - .map_err(|_err| FsError::GeneralFailure)?; - } - } - } - Ok(()) - }); - - Self { send } - } - - /// Get a [`Node`] from the specified [`DavPath`]. - /// - /// # Arguments - /// - /// * `path` - The path to get the [`Tree`] at - /// - /// # Errors - /// - /// * If the [`Tree`] could not be found - /// - /// # Returns - /// - /// The [`Node`] at the specified path - /// - /// [`Tree`]: crate::repofile::Tree - async fn node_from_path(&self, path: &DavPath) -> Result { - let (send, rcv) = oneshot::channel(); - self.send - .send(DavFsInnerCommand::Node(path.clone(), send)) - .await - .map_err(|_err| FsError::GeneralFailure)?; - rcv.await.map_err(|_err| FsError::GeneralFailure)? - } - - /// Get a list of [`Node`]s from the specified directory path. - /// - /// # Arguments - /// - /// * `path` - The path to get the [`Tree`] at - /// - /// # Errors - /// - /// * If the [`Tree`] could not be found - /// - /// # Returns - /// - /// The list of [`Node`]s at the specified path - /// - /// [`Tree`]: crate::repofile::Tree - async fn dir_entries_from_path(&self, path: &DavPath) -> Result, FsError> { - let (send, rcv) = oneshot::channel(); - self.send - .send(DavFsInnerCommand::DirEntries(path.clone(), send)) - .await - .map_err(|_err| FsError::GeneralFailure)?; - rcv.await.map_err(|_err| FsError::GeneralFailure)? - } - - async fn open(&self, node: &Node, options: OpenOptions) -> Result { - let (send, rcv) = oneshot::channel(); - self.send - .send(DavFsInnerCommand::Open(node.clone(), options, send)) - .await - .map_err(|_err| FsError::GeneralFailure)?; - rcv.await.map_err(|_err| FsError::GeneralFailure)? - } - async fn read_bytes( - &self, - file: OpenFile, - seek: usize, - count: usize, - ) -> Result<(Bytes, OpenFile), FsError> { - let (send, rcv) = oneshot::channel(); - self.send - .send(DavFsInnerCommand::ReadBytes(file, seek, count, send)) - .await - .map_err(|_err| FsError::GeneralFailure)?; - rcv.await.map_err(|_err| FsError::GeneralFailure)? - } -} - -impl DavFileSystem for WebDavFS { +impl DavFileSystem + for WebDavFS +{ fn metadata<'a>(&'a self, davpath: &'a DavPath) -> FsFuture<'_, Box> { self.symlink_metadata(davpath) } fn symlink_metadata<'a>(&'a self, davpath: &'a DavPath) -> FsFuture<'_, Box> { async move { - let node = self.node_from_path(davpath).await?; + let node = self.node_from_path(davpath)?; let meta: Box = Box::new(DavFsMetaData(node)); Ok(meta) } @@ -295,7 +154,7 @@ impl DavFileSystem for WebDavFS { _meta: ReadDirMeta, ) -> FsFuture<'_, FsStream>> { async move { - let entries = self.dir_entries_from_path(davpath).await?; + let entries = self.dir_entries_from_path(davpath)?; let entry_iter = entries.into_iter().map(|e| { let entry: Box = Box::new(DavFsDirEntry(e)); Ok(entry) @@ -312,13 +171,30 @@ impl DavFileSystem for WebDavFS { options: OpenOptions, ) -> FsFuture<'_, Box> { async move { - let node = self.node_from_path(path).await?; - let file = self.open(&node, options).await?; + if options.write + || options.append + || options.truncate + || options.create + || options.create_new + { + return Err(FsError::Forbidden); + } + + let node = self.node_from_path(path)?; + if matches!(self.inner.file_policy, FilePolicy::Forbidden) { + return Err(FsError::Forbidden); + } + + let open = self + .inner + .repo + .open_file(&node) + .map_err(|_err| FsError::GeneralFailure)?; let file: Box = Box::new(DavFsFile { - open: Some(file), - seek: 0, - fs: self.clone(), node, + open, + fs: self.inner.clone(), + seek: 0, }); Ok(file) } @@ -358,25 +234,27 @@ impl DavDirEntry for DavFsDirEntry { /// A [`DavFile`] implementation for [`Node`]s. /// /// This is a read-only file. -struct DavFsFile { +struct DavFsFile { /// The [`Node`] this file is for node: Node, /// The [`OpenFile`] for this file - open: Option, + open: OpenFile, + + /// The [`DavFsInner`] this file belongs to + fs: Arc>, /// The current seek position seek: usize, - fs: WebDavFS, } -impl Debug for DavFsFile { +impl Debug for DavFsFile { fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> { write!(f, "DavFile") } } -impl DavFile for DavFsFile { +impl DavFile for DavFsFile { fn metadata(&mut self) -> FsFuture<'_, Box> { async move { let meta: Box = Box::new(DavFsMetaData(self.node.clone())); @@ -395,16 +273,12 @@ impl DavFile for DavFsFile { fn read_bytes(&mut self, count: usize) -> FsFuture<'_, Bytes> { async move { - let (data, open) = self + let data = self .fs - .read_bytes( - self.open.take().ok_or(FsError::GeneralFailure)?, - self.seek, - count, - ) - .await?; + .repo + .read_file_at(&self.open, self.seek, count) + .map_err(|_err| FsError::GeneralFailure)?; self.seek += data.len(); - self.open = Some(open); Ok(data) } .boxed() @@ -418,7 +292,7 @@ impl DavFile for DavFsFile { } SeekFrom::Current(delta) => { self.seek = usize::try_from( - i64::try_from(self.seek).expect("i64 should not wrap around") + delta, + i64::try_from(self.seek).expect("i64 wrapped around") + delta, ) .expect("usize overflow should not happen"); }