From 92180316078c8ecb991fc0e5318e2d8c8b4ea231 Mon Sep 17 00:00:00 2001 From: InvalidJoker <0limitdev@gmail.com> Date: Fri, 16 Oct 2026 15:06:16 +0200 Subject: [PATCH 1/3] feat: batch methods Assisted-By: Claude Opus 5.5 --- daemon/src/files/batch.rs | 466 ++++++++++++++++++++++++ daemon/src/files/mod.rs | 2 + daemon/src/files/write.rs | 2 +- daemon/src/service/files.rs | 37 +- daemon_common/src/service/files.rs | 121 ++++++ daemon_common/src/service/files/task.rs | 2 + 6 files changed, 621 insertions(+), 9 deletions(-) create mode 100644 daemon/src/files/batch.rs diff --git a/daemon/src/files/batch.rs b/daemon/src/files/batch.rs new file mode 100644 index 00000000..3edd476a --- /dev/null +++ b/daemon/src/files/batch.rs @@ -0,0 +1,466 @@ +// SPDX-License-Identifier: AGPL-3.0-only +use crate::files::storage_path; +use crate::storage::StorageIdentifier; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; +use crate::utils::storage_parent_ino; +use crate::{Daemon, DaemonExtra, StorageNotification}; +use cap_async_std::fs::{Dir, MetadataExt as _, OpenOptions}; +use core::fmt::Write as _; +use daemon_common::service::error_storage_failure::FromStorageFailure; +use daemon_common::service::error_storage_mutex_locked::FromStorageMutexLocked; +use daemon_common::service::error_storage_not_found::FromStorageNotFound; +use daemon_common::service::files::task::{FileOperationKind, StorageTask}; +use daemon_common::service::files::{ + FileOperation, FileOperationBatch, FileOperationFailure, FileOperationResult, UrlWrite, + UrlWriteBatch, UrlWriteFailure, UrlWriteResult, +}; +use futures::StreamExt as _; +use sha2::{Digest as _, Sha512}; +use std::io; +use std::io::ErrorKind; +use std::os::fd::OwnedFd; +use std::path::{Path, PathBuf}; +use tokio::io::{AsyncWriteExt as _, BufWriter}; +use tokio::sync::broadcast::Sender; +use tracing::{debug, warn}; + +const DEFAULT_PARALLEL: u8 = 8; +const MAX_PARALLEL: u8 = 16; +const DEFAULT_ATTEMPTS: u8 = 3; +const MAX_ATTEMPTS: u8 = 5; +const PART_SUFFIX: &str = ".part"; + +fn url_failure(err: &io::Error) -> UrlWriteFailure { + match err.kind() { + ErrorKind::NotFound => UrlWriteFailure::NotFound, + ErrorKind::PermissionDenied => UrlWriteFailure::SandboxViolated, + ErrorKind::IsADirectory => UrlWriteFailure::IsDirectory, + ErrorKind::QuotaExceeded => UrlWriteFailure::QuotaReached, + _ => UrlWriteFailure::Io, + } +} + +fn operation_failure(err: &io::Error) -> FileOperationFailure { + match err.kind() { + ErrorKind::NotFound => FileOperationFailure::NotFound, + ErrorKind::AlreadyExists | ErrorKind::DirectoryNotEmpty => { + FileOperationFailure::AlreadyExists + } + ErrorKind::PermissionDenied => FileOperationFailure::SandboxViolated, + ErrorKind::IsADirectory => FileOperationFailure::IsDirectory, + ErrorKind::NotADirectory => FileOperationFailure::NotDirectory, + ErrorKind::QuotaExceeded => FileOperationFailure::QuotaReached, + _ => FileOperationFailure::Io, + } +} + +const fn retryable(failure: UrlWriteFailure) -> bool { + matches!( + failure, + UrlWriteFailure::RequestFailed + | UrlWriteFailure::NotSuccess + | UrlWriteFailure::ReadFailed + | UrlWriteFailure::SizeMismatch + | UrlWriteFailure::HashMismatch + | UrlWriteFailure::Io + ) +} + +fn hex(bytes: &[u8]) -> String { + let mut out = String::with_capacity(bytes.len().saturating_mul(2)); + for byte in bytes { + let _ = write!(out, "{byte:02x}"); + } + out +} + +fn part_path(path: &Path) -> PathBuf { + let mut part = path.as_os_str().to_owned(); + part.push(PART_SUFFIX); + PathBuf::from(part) +} + +async fn create_parents(storage: &Dir, path: &Path) -> Result<(), io::Error> { + let Some(parent) = path.parent().map(Path::to_path_buf) else { + return Ok(()); + }; + let storage = Dir::clone(storage); + tokio::task::spawn_blocking(move || storage.create_dir_all(&parent)) + .await + .map_err(io::Error::from)? +} + +fn notify_update(notify: &Sender, parent: Option, inode: Option) { + let _ = notify + .send(StorageNotification::Update { + parent_inode: parent, + inode, + }) + .inspect_err(|err| warn!("Unable to trigger storage notification: {err:?}")); +} + +async fn download_once( + daemon: &Daemon, + storage: &Dir, + part: &Path, + file: &UrlWrite, +) -> Result<(), UrlWriteFailure> { + let mut response = daemon + .client + .get(file.url.clone()) + .send() + .await + .inspect_err(|err| warn!("Batch download of {} failed (request): {err:?}", file.url)) + .map_err(|_err| UrlWriteFailure::RequestFailed)? + .error_for_status() + .inspect_err(|err| warn!("Batch download of {} failed (status): {err:?}", file.url)) + .map_err(|_err| UrlWriteFailure::NotSuccess)?; + + let fd: OwnedFd = storage + .open_with( + part, + OpenOptions::new().create(true).truncate(true).write(true), + ) + .await + .inspect_err(|err| debug!("Batch download could not open {part:?}: {err:?}")) + .map_err(|err| url_failure(&err))? + .into_std() + .into(); + let mut writer = BufWriter::new(tokio::fs::File::from_std(std::fs::File::from(fd))); + + let size_limit = file.size.filter(|_size| file.sha512.is_none()); + let mut hasher = Sha512::new(); + let mut written: u64 = 0; + while let Some(chunk) = response + .chunk() + .await + .inspect_err(|err| warn!("Batch download of {} failed (read): {err:?}", file.url)) + .map_err(|_err| UrlWriteFailure::ReadFailed)? + { + written = written.saturating_add(chunk.len() as u64); + if size_limit.is_some_and(|size| written > size) { + return Err(UrlWriteFailure::SizeMismatch); + } + hasher.update(&chunk); + writer + .write_all(&chunk) + .await + .map_err(|err| url_failure(&err))?; + } + writer.flush().await.map_err(|err| url_failure(&err))?; + drop(writer); + + if let Some(expected) = &file.sha512 { + if !hex(&hasher.finalize()).eq_ignore_ascii_case(expected) { + return Err(UrlWriteFailure::HashMismatch); + } + if file.size.is_some_and(|size| written != size) { + debug!( + "{} has {written} bytes instead of the declared size, but its hash matches", + file.url + ); + } + } else if size_limit.is_some_and(|size| written != size) { + return Err(UrlWriteFailure::SizeMismatch); + } else { + debug!( + "{} has no declared hash, only its size was checked", + file.url + ); + } + + Ok(()) +} + +async fn write_one( + daemon: &Daemon, + storage: Dir, + path: PathBuf, + file: &UrlWrite, + attempts: u8, + notify: &Sender, +) -> Result<(), UrlWriteFailure> { + create_parents(&storage, &path) + .await + .map_err(|err| url_failure(&err))?; + let part = part_path(&path); + + let mut last = UrlWriteFailure::Io; + for attempt in 1..=attempts { + match download_once(daemon, &storage, &part, file).await { + Ok(()) => { + storage + .rename(&part, &storage, &path) + .await + .map_err(|err| url_failure(&err))?; + let parent = storage_parent_ino(&storage, &path).await.ok(); + let inode = storage.metadata(&path).await.map(|meta| meta.ino()).ok(); + notify_update(notify, parent, inode); + return Ok(()); + } + Err(failure) => { + let _ = storage.remove_file(&part).await; + debug!("Batch download attempt {attempt} of {path:?} failed: {failure:?}"); + if !retryable(failure) { + return Err(failure); + } + last = failure; + } + } + } + + Err(last) +} + +/// Cancel safe +pub async fn write_files_by_url< + E: From + + From + + FromStorageNotFound + + FromStorageMutexLocked + + FromStorageFailure, +>( + daemon: &Daemon, + id: StorageIdentifier, + batch: UrlWriteBatch, +) -> Result, E> { + let StorageLockData { + lock, + storage, + notify, + tokens, + .. + } = storage_read_timeout::( + &daemon.config, + &*daemon.conn, + &daemon.storages, + id, + StorageTask::FileOperation { + kind: FileOperationKind::WriteByUrlBatch, + }, + ) + .await?; + + let parallel = usize::from( + batch + .parallel + .unwrap_or(DEFAULT_PARALLEL) + .clamp(1, MAX_PARALLEL), + ); + let attempts = batch + .attempts + .unwrap_or(DEFAULT_ATTEMPTS) + .clamp(1, MAX_ATTEMPTS); + + let (entry, token_map, sender) = (&storage, &tokens, ¬ify); + let results = futures::stream::iter(batch.files) + .map(|file| async move { + let error = match storage_path(id, entry, file.path.clone(), token_map).await { + Ok((dir, _, path)) => write_one(daemon, dir, path, &file, attempts, sender) + .await + .err(), + Err(err) => Some(url_failure(&err)), + }; + UrlWriteResult { + path: file.path, + error, + } + }) + .buffered(parallel) + .collect::>() + .await; + + drop(lock); + Ok(results) +} + +async fn resolve( + id: StorageIdentifier, + storage: &crate::config::StorageEntry, + tokens: &StorageTokens, + path: String, +) -> Result<(Dir, PathBuf), io::Error> { + let (dir, _, path) = storage_path(id, storage, path, tokens).await?; + Ok((dir, path)) +} + +async fn exists(storage: &Dir, path: &Path) -> bool { + storage.symlink_metadata(path).await.is_ok() +} + +async fn delete_path(storage: &Dir, path: PathBuf, recursive: bool) -> Result<(), io::Error> { + let metadata = match storage.symlink_metadata(&path).await { + Ok(metadata) => metadata, + Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()), + Err(err) => return Err(err), + }; + + if metadata.is_dir() { + if !recursive { + return Err(io::Error::from(ErrorKind::IsADirectory)); + } + let storage = Dir::clone(storage); + tokio::task::spawn(async move { storage.remove_dir_all(&path).await }) + .await + .map_err(io::Error::from)? + } else { + storage.remove_file(&path).await + } +} + +async fn move_path( + from_dir: &Dir, + from: &Path, + to_dir: &Dir, + to: &Path, + overwrite: bool, +) -> Result<(), io::Error> { + if !exists(from_dir, from).await { + return Err(io::Error::from(ErrorKind::NotFound)); + } + if exists(to_dir, to).await { + if !overwrite { + return Err(io::Error::from(ErrorKind::AlreadyExists)); + } + delete_path(to_dir, to.to_path_buf(), true).await?; + } + create_parents(to_dir, to).await?; + from_dir.rename(from, to_dir, to).await +} + +async fn copy_path( + from_dir: &Dir, + from: &Path, + to_dir: &Dir, + to: &Path, + overwrite: bool, +) -> Result<(), io::Error> { + let mut queue = vec![(from.to_path_buf(), to.to_path_buf())]; + + while let Some((source, target)) = queue.pop() { + let metadata = from_dir.symlink_metadata(&source).await?; + if metadata.is_symlink() { + continue; + } + + if metadata.is_dir() { + create_parents(to_dir, &target.join("_")).await?; + for entry in from_dir.read_dir(&source).await? { + let name = entry?.file_name(); + queue.push((source.join(&name), target.join(&name))); + } + continue; + } + + if exists(to_dir, &target).await { + if !overwrite { + return Err(io::Error::from(ErrorKind::AlreadyExists)); + } + delete_path(to_dir, target.clone(), false).await?; + } + create_parents(to_dir, &target).await?; + from_dir.copy(&source, to_dir, &target).await?; + } + + Ok(()) +} + +async fn run_operation( + id: StorageIdentifier, + storage: &crate::config::StorageEntry, + tokens: &StorageTokens, + notify: &Sender, + operation: FileOperation, +) -> Result<(), io::Error> { + let touched = match operation { + FileOperation::Delete { path, recursive } => { + let (dir, path) = resolve(id, storage, tokens, path).await?; + let parent = storage_parent_ino(&dir, &path).await.ok(); + delete_path(&dir, path, recursive).await?; + parent + } + FileOperation::Mkdir { path } => { + let (dir, path) = resolve(id, storage, tokens, path).await?; + create_parents(&dir, &path.join("_")).await?; + storage_parent_ino(&dir, &path).await.ok() + } + FileOperation::Move { + from, + to, + overwrite, + } => { + let (from_dir, from) = resolve(id, storage, tokens, from).await?; + let (to_dir, to) = resolve(id, storage, tokens, to).await?; + if let Ok(parent) = storage_parent_ino(&from_dir, &from).await { + notify_update(notify, Some(parent), None); + } + move_path(&from_dir, &from, &to_dir, &to, overwrite).await?; + storage_parent_ino(&to_dir, &to).await.ok() + } + FileOperation::Copy { + from, + to, + overwrite, + } => { + let (from_dir, from) = resolve(id, storage, tokens, from).await?; + let (to_dir, to) = resolve(id, storage, tokens, to).await?; + copy_path(&from_dir, &from, &to_dir, &to, overwrite).await?; + storage_parent_ino(&to_dir, &to).await.ok() + } + }; + + notify_update(notify, touched, None); + Ok(()) +} + +/// Cancel safe +pub async fn file_operations< + E: From + + From + + FromStorageNotFound + + FromStorageMutexLocked + + FromStorageFailure, +>( + daemon: &Daemon, + id: StorageIdentifier, + batch: FileOperationBatch, +) -> Result, E> { + let StorageLockData { + lock, + storage, + notify, + tokens, + .. + } = storage_read_timeout::( + &daemon.config, + &*daemon.conn, + &daemon.storages, + id, + StorageTask::FileOperation { + kind: FileOperationKind::Batch, + }, + ) + .await?; + + let mut results = Vec::with_capacity(batch.operations.len()); + let mut failed = false; + for operation in batch.operations { + if failed && batch.stop_on_error { + results.push(FileOperationResult { + error: Some(FileOperationFailure::Skipped), + }); + continue; + } + + let error = run_operation(id, &storage, &tokens, ¬ify, operation) + .await + .inspect_err(|err| debug!("Batch file operation failed: {err:?}")) + .err() + .map(|err| operation_failure(&err)); + failed |= error.is_some(); + results.push(FileOperationResult { error }); + } + + drop(lock); + Ok(results) +} diff --git a/daemon/src/files/mod.rs b/daemon/src/files/mod.rs index cdf8755f..f31f056b 100644 --- a/daemon/src/files/mod.rs +++ b/daemon/src/files/mod.rs @@ -1,5 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-only mod archive; +mod batch; mod chmod; mod chown; mod copy; @@ -19,6 +20,7 @@ use crate::storage::sync::StorageTokens; use crate::storage::{STORAGE_TOKEN_MAGIC, StorageIdentifier}; use crate::utils::path_relative; pub use archive::{create_archive, unpack_archive}; +pub use batch::{file_operations, write_files_by_url}; use cap_async_std::ambient_authority; use cap_async_std::fs::{Dir, FileTypeExt as _, Metadata, MetadataExt as _}; pub use chmod::chmod_file; diff --git a/daemon/src/files/write.rs b/daemon/src/files/write.rs index 9d8916f8..3adaf619 100644 --- a/daemon/src/files/write.rs +++ b/daemon/src/files/write.rs @@ -286,7 +286,7 @@ async fn write_file_by_url0< .map_err(|_err| ::from())? { writer - .write(&chunk) + .write_all(&chunk) .await .inspect_err(|err| match err.kind() { ErrorKind::QuotaExceeded => debug!("Failed to write file by url: {err:?}"), diff --git a/daemon/src/service/files.rs b/daemon/src/service/files.rs index d4be273a..c07f3e53 100644 --- a/daemon/src/service/files.rs +++ b/daemon/src/service/files.rs @@ -2,23 +2,26 @@ use std::collections::HashSet; // SPDX-License-Identifier: AGPL-3.0-only use crate::files::{ - chmod_file, chown_file, copy_file, create_archive, delete_file, download_file, list_files, - mkdir_file, rename_file, stat_file, symlink_file, unpack_archive, watch_files, wipe, - write_file, write_file_by_template, write_file_by_url, + chmod_file, chown_file, copy_file, create_archive, delete_file, download_file, file_operations, + list_files, mkdir_file, rename_file, stat_file, symlink_file, unpack_archive, watch_files, + wipe, write_file, write_file_by_template, write_file_by_url, write_files_by_url, }; use crate::storage::StorageIdentifier; use crate::{Daemon, DaemonExtra}; use async_trait::async_trait; use axum::extract::WebSocketUpgrade; -use daemon_common::service::files::FileEntry; use daemon_common::service::files::FilesService; use daemon_common::service::files::files_service_error::{ ServerChmodError, ServerChownError, ServerCopyFileError, ServerCreateArchiveError, ServerCreateDirectoryError, ServerDeleteFileError, ServerDownloadFileError, - ServerDownloadFileUserError, ServerListFilesError, ServerRenameFileError, ServerStatFileError, - ServerSymlinkError, ServerUnpackArchiveError, ServerUploadFileError, ServerUploadFileUserError, - ServerWatchFilesError, ServerWatchFilesUserError, ServerWriteFileByTemplateError, - ServerWriteFileByUrlError, WipeServerError, + ServerDownloadFileUserError, ServerFileOperationsError, ServerListFilesError, + ServerRenameFileError, ServerStatFileError, ServerSymlinkError, ServerUnpackArchiveError, + ServerUploadFileError, ServerUploadFileUserError, ServerWatchFilesError, + ServerWatchFilesUserError, ServerWriteFileByTemplateError, ServerWriteFileByUrlError, + ServerWriteFilesByUrlError, WipeServerError, +}; +use daemon_common::service::files::{ + FileEntry, FileOperationBatch, FileOperationResult, UrlWriteBatch, UrlWriteResult, }; use daemon_common::service::server::ServerFileWrite; use daemon_common::service::{DataStream, FileResponse, WebsocketResponse}; @@ -133,6 +136,24 @@ impl FilesService for Daemon { write_file_by_url(self, storage, path, url).await } + async fn server_write_files_by_url( + &self, + server_id: u32, + batch: UrlWriteBatch, + ) -> Result, ServerWriteFilesByUrlError> { + let storage = StorageIdentifier::Primary(server_id); + write_files_by_url(self, storage, batch).await + } + + async fn server_file_operations( + &self, + server_id: u32, + batch: FileOperationBatch, + ) -> Result, ServerFileOperationsError> { + let storage = StorageIdentifier::Primary(server_id); + file_operations(self, storage, batch).await + } + async fn server_write_file_by_template( &self, server_id: u32, diff --git a/daemon_common/src/service/files.rs b/daemon_common/src/service/files.rs index d59d00e9..12e4efae 100644 --- a/daemon_common/src/service/files.rs +++ b/daemon_common/src/service/files.rs @@ -226,6 +226,36 @@ pub trait FilesService { verify: UserSignature, ) -> Result; + /// Downloads many files from urls in one call + /// + /// The storage lock is taken once for the whole batch. Missing parent directories are + /// created and up to `parallel` files are downloaded at the same time, each retried up to + /// `attempts` times. A file is first written to `.part` and only moved to `path` + /// once it is verified, so it is either complete or absent. With `sha512` the hash decides + /// (a wrong declared `size` is tolerated), otherwise `size` must match when given. + /// Every file gets its own result, one failed download does not stop the others. + #[route("/url/batch", method = POST)] + #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] + async fn server_write_files_by_url( + &self, + server_id: u32, + batch: UrlWriteBatch, + ) -> Result, _>; + + /// Runs many file operations in one call + /// + /// The operations run in order under a single storage lock and each one gets its own + /// result at the same index. Deleting a missing path counts as success, moves and copies + /// create missing parent directories, and copying a directory merges it into the target. + /// With `stop_on_error` the remaining operations are skipped after the first failure. + #[route("/batch", method = POST)] + #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] + async fn server_file_operations( + &self, + server_id: u32, + batch: FileOperationBatch, + ) -> Result, _>; + /// Deletes all files #[route("/wipe", method = GET)] #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] @@ -346,3 +376,94 @@ pub enum FilesWatchResponseEntry { offset: i64, }, } + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct UrlWrite { + pub path: String, + #[schema(value_type = String)] + pub url: Url, + #[serde(default)] + pub size: Option, + #[serde(default)] + pub sha512: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct UrlWriteBatch { + pub files: Vec, + #[serde(default)] + pub parallel: Option, + #[serde(default)] + pub attempts: Option, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum UrlWriteFailure { + RequestFailed, + NotSuccess, + ReadFailed, + SizeMismatch, + HashMismatch, + QuotaReached, + NotFound, + SandboxViolated, + IsDirectory, + Io, +} + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct UrlWriteResult { + pub path: String, + pub error: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(tag = "op", rename_all = "snake_case")] +pub enum FileOperation { + Delete { + path: String, + #[serde(default)] + recursive: bool, + }, + Move { + from: String, + to: String, + #[serde(default)] + overwrite: bool, + }, + Copy { + from: String, + to: String, + #[serde(default)] + overwrite: bool, + }, + Mkdir { + path: String, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct FileOperationBatch { + pub operations: Vec, + #[serde(default)] + pub stop_on_error: bool, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum FileOperationFailure { + NotFound, + AlreadyExists, + SandboxViolated, + IsDirectory, + NotDirectory, + QuotaReached, + Skipped, + Io, +} + +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct FileOperationResult { + pub error: Option, +} diff --git a/daemon_common/src/service/files/task.rs b/daemon_common/src/service/files/task.rs index 946a0971..74e1d05d 100644 --- a/daemon_common/src/service/files/task.rs +++ b/daemon_common/src/service/files/task.rs @@ -64,7 +64,9 @@ pub enum FileOperationKind { Symlink, Write, WriteByUrl, + WriteByUrlBatch, WriteByTemplate, + Batch, } #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] From ae6a040ba971f5efc8bd2d7b0598aa030ccbe683 Mon Sep 17 00:00:00 2001 From: InvalidJoker <0limitdev@gmail.com> Date: Fri, 16 Oct 2026 16:00:48 +0200 Subject: [PATCH 2/3] fix: duplicated code --- daemon/src/files/batch.rs | 395 ++++------------------------- daemon/src/files/copy.rs | 23 +- daemon/src/files/delete.rs | 21 +- daemon/src/files/list.rs | 18 +- daemon/src/files/mkdir.rs | 21 +- daemon/src/files/rename.rs | 23 +- daemon/src/files/stat.rs | 18 +- daemon/src/files/write.rs | 34 ++- daemon_common/src/service/files.rs | 111 ++++---- 9 files changed, 250 insertions(+), 414 deletions(-) diff --git a/daemon/src/files/batch.rs b/daemon/src/files/batch.rs index 3edd476a..ed699c3d 100644 --- a/daemon/src/files/batch.rs +++ b/daemon/src/files/batch.rs @@ -1,216 +1,29 @@ // SPDX-License-Identifier: AGPL-3.0-only -use crate::files::storage_path; +use crate::config::StorageEntry; +use crate::files::copy::copy_file_locked; +use crate::files::delete::delete_file_locked; +use crate::files::list::list_files_locked; +use crate::files::mkdir::mkdir_file_locked; +use crate::files::rename::rename_file_locked; +use crate::files::stat::stat_file_locked; +use crate::files::write::write_file_by_url_locked; use crate::storage::StorageIdentifier; use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; -use crate::utils::storage_parent_ino; use crate::{Daemon, DaemonExtra, StorageNotification}; -use cap_async_std::fs::{Dir, MetadataExt as _, OpenOptions}; -use core::fmt::Write as _; use daemon_common::service::error_storage_failure::FromStorageFailure; use daemon_common::service::error_storage_mutex_locked::FromStorageMutexLocked; use daemon_common::service::error_storage_not_found::FromStorageNotFound; use daemon_common::service::files::task::{FileOperationKind, StorageTask}; use daemon_common::service::files::{ - FileOperation, FileOperationBatch, FileOperationFailure, FileOperationResult, UrlWrite, - UrlWriteBatch, UrlWriteFailure, UrlWriteResult, + BatchOperationError, FileOperation, FileOperationBatch, FileOperationFailure, + FileOperationResult, UrlWriteBatch, UrlWriteResult, }; use futures::StreamExt as _; -use sha2::{Digest as _, Sha512}; use std::io; -use std::io::ErrorKind; -use std::os::fd::OwnedFd; -use std::path::{Path, PathBuf}; -use tokio::io::{AsyncWriteExt as _, BufWriter}; use tokio::sync::broadcast::Sender; -use tracing::{debug, warn}; const DEFAULT_PARALLEL: u8 = 8; const MAX_PARALLEL: u8 = 16; -const DEFAULT_ATTEMPTS: u8 = 3; -const MAX_ATTEMPTS: u8 = 5; -const PART_SUFFIX: &str = ".part"; - -fn url_failure(err: &io::Error) -> UrlWriteFailure { - match err.kind() { - ErrorKind::NotFound => UrlWriteFailure::NotFound, - ErrorKind::PermissionDenied => UrlWriteFailure::SandboxViolated, - ErrorKind::IsADirectory => UrlWriteFailure::IsDirectory, - ErrorKind::QuotaExceeded => UrlWriteFailure::QuotaReached, - _ => UrlWriteFailure::Io, - } -} - -fn operation_failure(err: &io::Error) -> FileOperationFailure { - match err.kind() { - ErrorKind::NotFound => FileOperationFailure::NotFound, - ErrorKind::AlreadyExists | ErrorKind::DirectoryNotEmpty => { - FileOperationFailure::AlreadyExists - } - ErrorKind::PermissionDenied => FileOperationFailure::SandboxViolated, - ErrorKind::IsADirectory => FileOperationFailure::IsDirectory, - ErrorKind::NotADirectory => FileOperationFailure::NotDirectory, - ErrorKind::QuotaExceeded => FileOperationFailure::QuotaReached, - _ => FileOperationFailure::Io, - } -} - -const fn retryable(failure: UrlWriteFailure) -> bool { - matches!( - failure, - UrlWriteFailure::RequestFailed - | UrlWriteFailure::NotSuccess - | UrlWriteFailure::ReadFailed - | UrlWriteFailure::SizeMismatch - | UrlWriteFailure::HashMismatch - | UrlWriteFailure::Io - ) -} - -fn hex(bytes: &[u8]) -> String { - let mut out = String::with_capacity(bytes.len().saturating_mul(2)); - for byte in bytes { - let _ = write!(out, "{byte:02x}"); - } - out -} - -fn part_path(path: &Path) -> PathBuf { - let mut part = path.as_os_str().to_owned(); - part.push(PART_SUFFIX); - PathBuf::from(part) -} - -async fn create_parents(storage: &Dir, path: &Path) -> Result<(), io::Error> { - let Some(parent) = path.parent().map(Path::to_path_buf) else { - return Ok(()); - }; - let storage = Dir::clone(storage); - tokio::task::spawn_blocking(move || storage.create_dir_all(&parent)) - .await - .map_err(io::Error::from)? -} - -fn notify_update(notify: &Sender, parent: Option, inode: Option) { - let _ = notify - .send(StorageNotification::Update { - parent_inode: parent, - inode, - }) - .inspect_err(|err| warn!("Unable to trigger storage notification: {err:?}")); -} - -async fn download_once( - daemon: &Daemon, - storage: &Dir, - part: &Path, - file: &UrlWrite, -) -> Result<(), UrlWriteFailure> { - let mut response = daemon - .client - .get(file.url.clone()) - .send() - .await - .inspect_err(|err| warn!("Batch download of {} failed (request): {err:?}", file.url)) - .map_err(|_err| UrlWriteFailure::RequestFailed)? - .error_for_status() - .inspect_err(|err| warn!("Batch download of {} failed (status): {err:?}", file.url)) - .map_err(|_err| UrlWriteFailure::NotSuccess)?; - - let fd: OwnedFd = storage - .open_with( - part, - OpenOptions::new().create(true).truncate(true).write(true), - ) - .await - .inspect_err(|err| debug!("Batch download could not open {part:?}: {err:?}")) - .map_err(|err| url_failure(&err))? - .into_std() - .into(); - let mut writer = BufWriter::new(tokio::fs::File::from_std(std::fs::File::from(fd))); - - let size_limit = file.size.filter(|_size| file.sha512.is_none()); - let mut hasher = Sha512::new(); - let mut written: u64 = 0; - while let Some(chunk) = response - .chunk() - .await - .inspect_err(|err| warn!("Batch download of {} failed (read): {err:?}", file.url)) - .map_err(|_err| UrlWriteFailure::ReadFailed)? - { - written = written.saturating_add(chunk.len() as u64); - if size_limit.is_some_and(|size| written > size) { - return Err(UrlWriteFailure::SizeMismatch); - } - hasher.update(&chunk); - writer - .write_all(&chunk) - .await - .map_err(|err| url_failure(&err))?; - } - writer.flush().await.map_err(|err| url_failure(&err))?; - drop(writer); - - if let Some(expected) = &file.sha512 { - if !hex(&hasher.finalize()).eq_ignore_ascii_case(expected) { - return Err(UrlWriteFailure::HashMismatch); - } - if file.size.is_some_and(|size| written != size) { - debug!( - "{} has {written} bytes instead of the declared size, but its hash matches", - file.url - ); - } - } else if size_limit.is_some_and(|size| written != size) { - return Err(UrlWriteFailure::SizeMismatch); - } else { - debug!( - "{} has no declared hash, only its size was checked", - file.url - ); - } - - Ok(()) -} - -async fn write_one( - daemon: &Daemon, - storage: Dir, - path: PathBuf, - file: &UrlWrite, - attempts: u8, - notify: &Sender, -) -> Result<(), UrlWriteFailure> { - create_parents(&storage, &path) - .await - .map_err(|err| url_failure(&err))?; - let part = part_path(&path); - - let mut last = UrlWriteFailure::Io; - for attempt in 1..=attempts { - match download_once(daemon, &storage, &part, file).await { - Ok(()) => { - storage - .rename(&part, &storage, &path) - .await - .map_err(|err| url_failure(&err))?; - let parent = storage_parent_ino(&storage, &path).await.ok(); - let inode = storage.metadata(&path).await.map(|meta| meta.ino()).ok(); - notify_update(notify, parent, inode); - return Ok(()); - } - Err(failure) => { - let _ = storage.remove_file(&part).await; - debug!("Batch download attempt {attempt} of {path:?} failed: {failure:?}"); - if !retryable(failure) { - return Err(failure); - } - last = failure; - } - } - } - - Err(last) -} /// Cancel safe pub async fn write_files_by_url< @@ -247,20 +60,21 @@ pub async fn write_files_by_url< .unwrap_or(DEFAULT_PARALLEL) .clamp(1, MAX_PARALLEL), ); - let attempts = batch - .attempts - .unwrap_or(DEFAULT_ATTEMPTS) - .clamp(1, MAX_ATTEMPTS); - let (entry, token_map, sender) = (&storage, &tokens, ¬ify); let results = futures::stream::iter(batch.files) .map(|file| async move { - let error = match storage_path(id, entry, file.path.clone(), token_map).await { - Ok((dir, _, path)) => write_one(daemon, dir, path, &file, attempts, sender) - .await - .err(), - Err(err) => Some(url_failure(&err)), - }; + let error = write_file_by_url_locked::( + daemon, + id, + entry, + token_map, + sender, + file.path.clone(), + file.url, + ) + .await + .err() + .map(FileOperationFailure::from); UrlWriteResult { path: file.path, error, @@ -274,143 +88,40 @@ pub async fn write_files_by_url< Ok(results) } -async fn resolve( - id: StorageIdentifier, - storage: &crate::config::StorageEntry, - tokens: &StorageTokens, - path: String, -) -> Result<(Dir, PathBuf), io::Error> { - let (dir, _, path) = storage_path(id, storage, path, tokens).await?; - Ok((dir, path)) -} - -async fn exists(storage: &Dir, path: &Path) -> bool { - storage.symlink_metadata(path).await.is_ok() -} - -async fn delete_path(storage: &Dir, path: PathBuf, recursive: bool) -> Result<(), io::Error> { - let metadata = match storage.symlink_metadata(&path).await { - Ok(metadata) => metadata, - Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()), - Err(err) => return Err(err), - }; - - if metadata.is_dir() { - if !recursive { - return Err(io::Error::from(ErrorKind::IsADirectory)); - } - let storage = Dir::clone(storage); - tokio::task::spawn(async move { storage.remove_dir_all(&path).await }) - .await - .map_err(io::Error::from)? - } else { - storage.remove_file(&path).await - } -} - -async fn move_path( - from_dir: &Dir, - from: &Path, - to_dir: &Dir, - to: &Path, - overwrite: bool, -) -> Result<(), io::Error> { - if !exists(from_dir, from).await { - return Err(io::Error::from(ErrorKind::NotFound)); - } - if exists(to_dir, to).await { - if !overwrite { - return Err(io::Error::from(ErrorKind::AlreadyExists)); - } - delete_path(to_dir, to.to_path_buf(), true).await?; - } - create_parents(to_dir, to).await?; - from_dir.rename(from, to_dir, to).await -} - -async fn copy_path( - from_dir: &Dir, - from: &Path, - to_dir: &Dir, - to: &Path, - overwrite: bool, -) -> Result<(), io::Error> { - let mut queue = vec![(from.to_path_buf(), to.to_path_buf())]; - - while let Some((source, target)) = queue.pop() { - let metadata = from_dir.symlink_metadata(&source).await?; - if metadata.is_symlink() { - continue; - } - - if metadata.is_dir() { - create_parents(to_dir, &target.join("_")).await?; - for entry in from_dir.read_dir(&source).await? { - let name = entry?.file_name(); - queue.push((source.join(&name), target.join(&name))); - } - continue; - } - - if exists(to_dir, &target).await { - if !overwrite { - return Err(io::Error::from(ErrorKind::AlreadyExists)); - } - delete_path(to_dir, target.clone(), false).await?; - } - create_parents(to_dir, &target).await?; - from_dir.copy(&source, to_dir, &target).await?; - } - - Ok(()) -} - async fn run_operation( id: StorageIdentifier, - storage: &crate::config::StorageEntry, + storage: &StorageEntry, tokens: &StorageTokens, notify: &Sender, operation: FileOperation, -) -> Result<(), io::Error> { - let touched = match operation { +) -> Result { + let mut result = FileOperationResult::default(); + match operation { + FileOperation::List { path } => { + result.entries = + Some(list_files_locked::(id, storage, tokens, path).await?); + } + FileOperation::Stat { path } => { + result.entry = + Some(stat_file_locked::(id, storage, tokens, path).await?); + } FileOperation::Delete { path, recursive } => { - let (dir, path) = resolve(id, storage, tokens, path).await?; - let parent = storage_parent_ino(&dir, &path).await.ok(); - delete_path(&dir, path, recursive).await?; - parent + delete_file_locked::(id, storage, tokens, notify, path, recursive) + .await?; } - FileOperation::Mkdir { path } => { - let (dir, path) = resolve(id, storage, tokens, path).await?; - create_parents(&dir, &path.join("_")).await?; - storage_parent_ino(&dir, &path).await.ok() + FileOperation::Rename { from, to } => { + rename_file_locked::(id, storage, tokens, notify, from, to) + .await?; } - FileOperation::Move { - from, - to, - overwrite, - } => { - let (from_dir, from) = resolve(id, storage, tokens, from).await?; - let (to_dir, to) = resolve(id, storage, tokens, to).await?; - if let Ok(parent) = storage_parent_ino(&from_dir, &from).await { - notify_update(notify, Some(parent), None); - } - move_path(&from_dir, &from, &to_dir, &to, overwrite).await?; - storage_parent_ino(&to_dir, &to).await.ok() + FileOperation::Copy { from, to } => { + copy_file_locked::(id, storage, tokens, notify, from, to).await?; } - FileOperation::Copy { - from, - to, - overwrite, - } => { - let (from_dir, from) = resolve(id, storage, tokens, from).await?; - let (to_dir, to) = resolve(id, storage, tokens, to).await?; - copy_path(&from_dir, &from, &to_dir, &to, overwrite).await?; - storage_parent_ino(&to_dir, &to).await.ok() + FileOperation::Mkdir { path, recursive } => { + mkdir_file_locked::(id, storage, tokens, notify, path, recursive) + .await?; } - }; - - notify_update(notify, touched, None); - Ok(()) + } + Ok(result) } /// Cancel safe @@ -448,17 +159,19 @@ pub async fn file_operations< if failed && batch.stop_on_error { results.push(FileOperationResult { error: Some(FileOperationFailure::Skipped), + ..FileOperationResult::default() }); continue; } - let error = run_operation(id, &storage, &tokens, ¬ify, operation) + let result = run_operation(id, &storage, &tokens, ¬ify, operation) .await - .inspect_err(|err| debug!("Batch file operation failed: {err:?}")) - .err() - .map(|err| operation_failure(&err)); - failed |= error.is_some(); - results.push(FileOperationResult { error }); + .unwrap_or_else(|err| FileOperationResult { + error: Some(err.into()), + ..FileOperationResult::default() + }); + failed |= result.error.is_some(); + results.push(result); } drop(lock); diff --git a/daemon/src/files/copy.rs b/daemon/src/files/copy.rs index 6ee4b58d..deaea68c 100644 --- a/daemon/src/files/copy.rs +++ b/daemon/src/files/copy.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::storage_path; use crate::storage::StorageIdentifier; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::utils::storage_parent_ino; use crate::{Daemon, DaemonExtra, StorageNotification}; use cap_async_std::fs::MetadataExt as _; @@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound; use daemon_common::service::files::task::{FileOperationKind, StorageTask}; use std::io; use std::io::ErrorKind; +use tokio::sync::broadcast::Sender; use tracing::{debug, error, warn}; /// Cancel safe @@ -43,8 +45,22 @@ pub async fn copy_file< }, ) .await?; - let (storage_from, _, path_from) = storage_path(id, &storage, from, &tokens).await?; - let (storage_to, _, path_to) = storage_path(id, &storage, to, &tokens).await?; + let result = copy_file_locked::(id, &storage, &tokens, ¬ify, from, to).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn copy_file_locked>( + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + notify: &Sender, + from: String, + to: String, +) -> Result<(), E> { + let (storage_from, _, path_from) = storage_path(id, storage, from, tokens).await?; + let (storage_to, _, path_to) = storage_path(id, storage, to, tokens).await?; let to_parent_ino = storage_parent_ino(&storage_to, &path_to).await; @@ -77,6 +93,5 @@ pub async fn copy_file< }); } - drop(lock); Ok(()) } diff --git a/daemon/src/files/delete.rs b/daemon/src/files/delete.rs index 8854f9e1..89d68e83 100644 --- a/daemon/src/files/delete.rs +++ b/daemon/src/files/delete.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::storage_path; use crate::storage::StorageIdentifier; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::utils::storage_parent_ino; use crate::{Daemon, DaemonExtra, StorageNotification}; use cap_async_std::fs::MetadataExt as _; @@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound; use daemon_common::service::files::task::{FileOperationKind, StorageTask}; use std::io; use std::io::ErrorKind; +use tokio::sync::broadcast::Sender; use tracing::{debug, error, warn}; /// Cancel safe @@ -43,7 +45,21 @@ pub async fn delete_file< }, ) .await?; - let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?; + let result = delete_file_locked::(id, &storage, &tokens, ¬ify, path, recursive).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn delete_file_locked>( + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + notify: &Sender, + path: String, + recursive: bool, +) -> Result<(), E> { + let (storage, _, path) = storage_path(id, storage, path, tokens).await?; let ino = storage.metadata(&path).await.map(|meta| meta.ino()); let parent_ino = storage_parent_ino(&storage, &path).await; @@ -80,6 +96,5 @@ pub async fn delete_file< }); } - drop(lock); Ok(()) } diff --git a/daemon/src/files/list.rs b/daemon/src/files/list.rs index 152a6744..1a38ea25 100644 --- a/daemon/src/files/list.rs +++ b/daemon/src/files/list.rs @@ -1,8 +1,9 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::{IntoFileEntry as _, storage_path}; use crate::storage::StorageIdentifier; use crate::storage::archive::format::ArchiveFormat; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::{Daemon, DaemonExtra}; use daemon_common::service::error_storage_failure::FromStorageFailure; use daemon_common::service::error_storage_mutex_locked::FromStorageMutexLocked; @@ -43,7 +44,19 @@ pub async fn list_files< }, ) .await?; - let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?; + let result = list_files_locked::(id, &storage, &tokens, path).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn list_files_locked>( + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + path: String, +) -> Result, E> { + let (storage, _, path) = storage_path(id, storage, path, tokens).await?; let read_dir = storage .read_dir(&path) .await @@ -97,6 +110,5 @@ pub async fn list_files< archive, )); } - drop(lock); Ok(files) } diff --git a/daemon/src/files/mkdir.rs b/daemon/src/files/mkdir.rs index 837f7e2a..f170ab10 100644 --- a/daemon/src/files/mkdir.rs +++ b/daemon/src/files/mkdir.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::storage_path; use crate::storage::StorageIdentifier; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::utils::storage_parent_ino; use crate::{Daemon, DaemonExtra, StorageNotification}; use cap_async_std::fs::MetadataExt as _; @@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound; use daemon_common::service::files::task::{FileOperationKind, StorageTask}; use std::io; use std::io::ErrorKind; +use tokio::sync::broadcast::Sender; use tracing::{debug, error, warn}; /// Cancel safe @@ -43,7 +45,21 @@ pub async fn mkdir_file< }, ) .await?; - let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?; + let result = mkdir_file_locked::(id, &storage, &tokens, ¬ify, path, recursive).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn mkdir_file_locked>( + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + notify: &Sender, + path: String, + recursive: bool, +) -> Result<(), E> { + let (storage, _, path) = storage_path(id, storage, path, tokens).await?; let parent_ino = storage_parent_ino(&storage, &path).await; @@ -85,6 +101,5 @@ pub async fn mkdir_file< }); } - drop(lock); Ok(()) } diff --git a/daemon/src/files/rename.rs b/daemon/src/files/rename.rs index 771e4543..51a26a5c 100644 --- a/daemon/src/files/rename.rs +++ b/daemon/src/files/rename.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::storage_path; use crate::storage::StorageIdentifier; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::utils::storage_parent_ino; use crate::{Daemon, DaemonExtra, StorageNotification}; use cap_async_std::fs::MetadataExt as _; @@ -11,6 +12,7 @@ use daemon_common::service::error_storage_not_found::FromStorageNotFound; use daemon_common::service::files::task::{FileOperationKind, StorageTask}; use std::io; use std::io::ErrorKind; +use tokio::sync::broadcast::Sender; use tracing::{debug, error, warn}; /// Cancel safe @@ -43,8 +45,22 @@ pub async fn rename_file< }, ) .await?; - let (storage_from, _, path_from) = storage_path(id, &storage, from, &tokens).await?; - let (storage_to, _, path_to) = storage_path(id, &storage, to, &tokens).await?; + let result = rename_file_locked::(id, &storage, &tokens, ¬ify, from, to).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn rename_file_locked>( + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + notify: &Sender, + from: String, + to: String, +) -> Result<(), E> { + let (storage_from, _, path_from) = storage_path(id, storage, from, tokens).await?; + let (storage_to, _, path_to) = storage_path(id, storage, to, tokens).await?; let from_parent_ino = storage_parent_ino(&storage_from, &path_from).await; let to_parent_ino = storage_parent_ino(&storage_to, &path_to).await; @@ -84,6 +100,5 @@ pub async fn rename_file< warn!("Rename was successful, but unable to trigger storage notification: {err:?}"); }); } - drop(lock); Ok(()) } diff --git a/daemon/src/files/stat.rs b/daemon/src/files/stat.rs index c08d8d0c..e6e26227 100644 --- a/daemon/src/files/stat.rs +++ b/daemon/src/files/stat.rs @@ -1,8 +1,9 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::{IntoFileEntry as _, storage_path}; use crate::storage::StorageIdentifier; use crate::storage::archive::format::ArchiveFormat; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::{Daemon, DaemonExtra}; use daemon_common::service::error_storage_failure::FromStorageFailure; use daemon_common::service::error_storage_mutex_locked::FromStorageMutexLocked; @@ -44,7 +45,19 @@ pub async fn stat_file< }, ) .await?; - let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?; + let result = stat_file_locked::(id, &storage, &tokens, path).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn stat_file_locked + FromFileNotFound>( + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + path: String, +) -> Result { + let (storage, _, path) = storage_path(id, storage, path, tokens).await?; let metadata = storage .symlink_metadata(&path) .await @@ -93,7 +106,6 @@ pub async fn stat_file< } let relative_path = path.join(filename.clone()); - drop(lock); Ok(metadata.into_fileentry( filename, relative_path.to_string_lossy().to_string(), diff --git a/daemon/src/files/write.rs b/daemon/src/files/write.rs index 3adaf619..4f4844a5 100644 --- a/daemon/src/files/write.rs +++ b/daemon/src/files/write.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: AGPL-3.0-only +use crate::config::StorageEntry; use crate::files::storage_path; use crate::storage::StorageIdentifier; -use crate::storage::sync::{StorageLockData, storage_read_timeout}; +use crate::storage::sync::{StorageLockData, StorageTokens, storage_read_timeout}; use crate::utils::storage_parent_ino; use crate::{Daemon, DaemonExtra, StorageNotification}; use cap_async_std::fs::{Dir, MetadataExt as _, OpenOptions}; @@ -223,18 +224,39 @@ pub async fn write_file_by_url< }, ) .await?; - let (storage, _, path) = storage_path(id, &storage, path, &tokens).await?; + let result = + write_file_by_url_locked::(daemon, id, &storage, &tokens, ¬ify, path, url).await; + drop(lock); + result +} + +/// Cancel safe, the caller holds the storage lock +pub(super) async fn write_file_by_url_locked< + E: From + + FromHttpRequestFailed + + FromHttpRequestNotSuccess + + FromHttpRequestReadFailed + + Send + + 'static, +>( + daemon: &Daemon, + id: StorageIdentifier, + storage: &StorageEntry, + tokens: &StorageTokens, + notify: &Sender, + path: String, + url: Url, +) -> Result<(), E> { + let (storage, _, path) = storage_path(id, storage, path, tokens).await?; tokio::task::spawn(write_file_by_url0::( daemon.clone(), storage, path, - notify, + notify.clone(), url, )) .await - .map_err(>::from)??; - drop(lock); - Ok(()) + .map_err(>::from)? } #[inline(always)] diff --git a/daemon_common/src/service/files.rs b/daemon_common/src/service/files.rs index 12e4efae..65ed1431 100644 --- a/daemon_common/src/service/files.rs +++ b/daemon_common/src/service/files.rs @@ -226,14 +226,11 @@ pub trait FilesService { verify: UserSignature, ) -> Result; - /// Downloads many files from urls in one call + /// Writes many files from urls in one call /// - /// The storage lock is taken once for the whole batch. Missing parent directories are - /// created and up to `parallel` files are downloaded at the same time, each retried up to - /// `attempts` times. A file is first written to `.part` and only moved to `path` - /// once it is verified, so it is either complete or absent. With `sha512` the hash decides - /// (a wrong declared `size` is tolerated), otherwise `size` must match when given. - /// Every file gets its own result, one failed download does not stop the others. + /// Takes the storage lock once and runs the same write as `GET /url` for every file, up to + /// `parallel` at a time. Every file gets its own result at the same index, one failed + /// download does not stop the others. #[route("/url/batch", method = POST)] #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] async fn server_write_files_by_url( @@ -244,10 +241,10 @@ pub trait FilesService { /// Runs many file operations in one call /// - /// The operations run in order under a single storage lock and each one gets its own - /// result at the same index. Deleting a missing path counts as success, moves and copies - /// create missing parent directories, and copying a directory merges it into the target. - /// With `stop_on_error` the remaining operations are skipped after the first failure. + /// Takes the storage lock once and runs the operations in order, each one exactly like its + /// own route (`/list`, `/stat`, `DELETE /file`, `/rename`, `/copy`, `/mkdir`). Every + /// operation gets its own result at the same index. With `stop_on_error` the remaining + /// operations are skipped after the first failure. #[route("/batch", method = POST)] #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] async fn server_file_operations( @@ -377,15 +374,24 @@ pub enum FilesWatchResponseEntry { }, } +error!( + BatchOperationError, + storage::{failure <- [io:*], quota_reached <- [io:QuotaExceeded]}, + file::{ + not_found <- [io:NotFound], + sandbox_violated <- [io:PermissionDenied], + already_exists <- [io:AlreadyExists], + not_directory <- [io:NotADirectory], + is_directory <- [io:IsADirectory] + }, + http_request::{failed, not_success, read_failed} +); + #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] pub struct UrlWrite { pub path: String, #[schema(value_type = String)] pub url: Url, - #[serde(default)] - pub size: Option, - #[serde(default)] - pub sha512: Option, } #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] @@ -393,53 +399,34 @@ pub struct UrlWriteBatch { pub files: Vec, #[serde(default)] pub parallel: Option, - #[serde(default)] - pub attempts: Option, -} - -#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] -#[serde(rename_all = "snake_case")] -pub enum UrlWriteFailure { - RequestFailed, - NotSuccess, - ReadFailed, - SizeMismatch, - HashMismatch, - QuotaReached, - NotFound, - SandboxViolated, - IsDirectory, - Io, -} - -#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] -pub struct UrlWriteResult { - pub path: String, - pub error: Option, } #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] #[serde(tag = "op", rename_all = "snake_case")] pub enum FileOperation { + List { + path: String, + }, + Stat { + path: String, + }, Delete { path: String, #[serde(default)] recursive: bool, }, - Move { + Rename { from: String, to: String, - #[serde(default)] - overwrite: bool, }, Copy { from: String, to: String, - #[serde(default)] - overwrite: bool, }, Mkdir { path: String, + #[serde(default)] + recursive: bool, }, } @@ -453,17 +440,47 @@ pub struct FileOperationBatch { #[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(rename_all = "snake_case")] pub enum FileOperationFailure { + StorageFailure, + QuotaReached, NotFound, - AlreadyExists, SandboxViolated, - IsDirectory, + AlreadyExists, NotDirectory, - QuotaReached, + IsDirectory, + HttpRequestFailed, + HttpRequestNotSuccess, + HttpRequestReadFailed, Skipped, - Io, +} + +impl From for FileOperationFailure { + fn from(value: BatchOperationError) -> Self { + match value { + BatchOperationError::StorageFailure | BatchOperationError::Communication(_) => { + Self::StorageFailure + } + BatchOperationError::StorageQuotaReached => Self::QuotaReached, + BatchOperationError::FileNotFound => Self::NotFound, + BatchOperationError::FileSandboxViolated => Self::SandboxViolated, + BatchOperationError::FileAlreadyExists => Self::AlreadyExists, + BatchOperationError::FileNotDirectory => Self::NotDirectory, + BatchOperationError::FileIsDirectory => Self::IsDirectory, + BatchOperationError::HttpRequestFailed => Self::HttpRequestFailed, + BatchOperationError::HttpRequestNotSuccess => Self::HttpRequestNotSuccess, + BatchOperationError::HttpRequestReadFailed => Self::HttpRequestReadFailed, + } + } } #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct UrlWriteResult { + pub path: String, + pub error: Option, +} + +#[derive(Clone, Default, Serialize, Deserialize, utoipa::ToSchema)] pub struct FileOperationResult { pub error: Option, + pub entry: Option, + pub entries: Option>, } From b7bef846eb31f8851ce432a9971b0dc61771a34a Mon Sep 17 00:00:00 2001 From: InvalidJoker <0limitdev@gmail.com> Date: Fri, 16 Oct 2026 16:01:34 +0200 Subject: [PATCH 3/3] docs: improve docs --- daemon_common/src/service/files.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/daemon_common/src/service/files.rs b/daemon_common/src/service/files.rs index 65ed1431..c9dc1723 100644 --- a/daemon_common/src/service/files.rs +++ b/daemon_common/src/service/files.rs @@ -228,9 +228,9 @@ pub trait FilesService { /// Writes many files from urls in one call /// - /// Takes the storage lock once and runs the same write as `GET /url` for every file, up to - /// `parallel` at a time. Every file gets its own result at the same index, one failed - /// download does not stop the others. + /// Runs the same write as `GET /url` for every file, up to `parallel` at a time, so a + /// caller with many files needs one request instead of one per file. Every file gets its + /// own result at the same index, one failed download does not stop the others. #[route("/url/batch", method = POST)] #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] async fn server_write_files_by_url( @@ -241,10 +241,11 @@ pub trait FilesService { /// Runs many file operations in one call /// - /// Takes the storage lock once and runs the operations in order, each one exactly like its - /// own route (`/list`, `/stat`, `DELETE /file`, `/rename`, `/copy`, `/mkdir`). Every - /// operation gets its own result at the same index. With `stop_on_error` the remaining - /// operations are skipped after the first failure. + /// Runs the operations in order, each one exactly like its own route (`/list`, `/stat`, + /// `DELETE /file`, `/rename`, `/copy`, `/mkdir`), so a caller with many operations needs + /// one request instead of one per operation. Every operation gets its own result at the + /// same index. With `stop_on_error` the remaining operations are skipped after the first + /// failure. #[route("/batch", method = POST)] #[error(storage::{not_found, mutex_locked, failure <- [io:*]}, sql::failure <- [sql:*])] async fn server_file_operations(