diff --git a/Cargo.lock b/Cargo.lock index a510da6e..0cf1901b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -278,6 +278,7 @@ dependencies = [ "cfg-if", "constant_time_eq", "cpufeatures", + "rayon-core", ] [[package]] @@ -1804,7 +1805,6 @@ checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" name = "oxicloud" version = "0.5.0" dependencies = [ - "anyhow", "argon2", "async-compression", "async-stream", diff --git a/Cargo.toml b/Cargo.toml index e1527e0a..65d736d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,7 +29,6 @@ uuid = { version = "1.21.0", features = ["v4", "serde"] } thiserror = "2.0.18" mockall = { version = "0.14.0", optional = true } sqlx = { version = "0.8.6", features = ["postgres", "runtime-tokio", "tls-rustls", "chrono", "uuid", "json"] } -anyhow = "1.0.102" jsonwebtoken = { version = "10.3.0", features = ["rust_crypto"] } argon2 = "0.5.3" rand_core = { version = "0.6", features = ["std", "getrandom"] } @@ -41,7 +40,7 @@ http-range-header = "0.4" image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] } md5 = "0.8.0" sha2 = "0.10.9" -blake3 = "1.8.3" +blake3 = { version = "1.8.3", features = ["rayon"] } hex = "0.4.3" http-body-util = "0.1.3" percent-encoding = "2.3" diff --git a/src/common/di.rs b/src/common/di.rs index 1f48d756..4ddf3340 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -549,6 +549,7 @@ impl AppServiceFactory { device_auth_service: None, app_password_service: None, path_resolver: None, + webdav_lock_store: crate::infrastructure::services::webdav_lock_service::create_webdav_lock_store(), }; // 9b. Wire admin settings service when auth is available @@ -854,6 +855,8 @@ pub struct AppState { Option>, pub path_resolver: Option>, + pub webdav_lock_store: + Arc, } // All AppState construction is done via struct literal in build_app_state(). diff --git a/src/infrastructure/auth_factory.rs b/src/infrastructure/auth_factory.rs index 9b9fbf2f..8caa422c 100644 --- a/src/infrastructure/auth_factory.rs +++ b/src/infrastructure/auth_factory.rs @@ -1,7 +1,8 @@ -use anyhow::Result; use sqlx::PgPool; use std::sync::Arc; +type Result = std::result::Result>; + use crate::application::services::auth_application_service::AuthApplicationService; use crate::application::services::folder_service::FolderService; use crate::common::config::AppConfig; diff --git a/src/infrastructure/db.rs b/src/infrastructure/db.rs index f153caae..af88b24d 100644 --- a/src/infrastructure/db.rs +++ b/src/infrastructure/db.rs @@ -1,8 +1,14 @@ use crate::common::config::AppConfig; -use anyhow::Result; use sqlx::{PgPool, postgres::PgPoolOptions}; use std::time::Duration; +/// Database initialization error. +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +pub struct DbError(String); + +type Result = std::result::Result; + /// Segmented database pools. /// /// `primary` is used for all user-facing request paths (REST, WebDAV, CalDAV, @@ -45,11 +51,11 @@ pub async fn create_database_pools(config: &AppConfig) -> Result { // Apply schema through the primary pool (idempotent) tracing::info!("Applying database schema..."); if let Err(e) = apply_schema(&primary).await { - return Err(anyhow::anyhow!( + return Err(DbError(format!( "Database schema could not be applied: {}. \ Run manually: psql -f db/schema.sql", e - )); + ))); } tracing::info!("Database schema applied successfully"); @@ -118,11 +124,11 @@ async fn create_pool_with_retries( Err(e) => { tracing::error!("Error verifying {} pool connection: {}", label, e); if attempt >= MAX_ATTEMPTS { - return Err(anyhow::anyhow!( + return Err(DbError(format!( "Error verifying PostgreSQL {} pool connection: {}", label, e - )); + ))); } } }, @@ -135,22 +141,22 @@ async fn create_pool_with_retries( e ); if attempt >= MAX_ATTEMPTS { - return Err(anyhow::anyhow!( + return Err(DbError(format!( "Error in PostgreSQL {} pool connection: {}", label, e - )); + ))); } tokio::time::sleep(Duration::from_secs(2)).await; } } } - Err(anyhow::anyhow!( + Err(DbError(format!( "Could not establish PostgreSQL {} pool connection after {} attempts", label, MAX_ATTEMPTS - )) + ))) } /// Apply the embedded schema.sql to the database. @@ -189,7 +195,7 @@ async fn apply_schema(pool: &PgPool) -> Result<()> { e, preview ); - return Err(anyhow::anyhow!("Schema statement {} failed: {}", i + 1, e)); + return Err(DbError(format!("Schema statement {} failed: {}", i + 1, e))); } } diff --git a/src/infrastructure/services/chunked_upload_service.rs b/src/infrastructure/services/chunked_upload_service.rs index e7c08c8e..cfd94491 100644 --- a/src/infrastructure/services/chunked_upload_service.rs +++ b/src/infrastructure/services/chunked_upload_service.rs @@ -660,6 +660,10 @@ impl ChunkedUploadService { let mut output = StdBufWriter::with_capacity(524_288, raw_output); let mut hasher = blake3::Hasher::new(); + // For files >10 MB, use multithreaded BLAKE3 hashing (all cores) + const RAYON_THRESHOLD: u64 = 10 * 1024 * 1024; + let use_rayon = total_size > RAYON_THRESHOLD; + // Single 512 KB read buffer reused across all chunks (avoids N allocations) let mut buf = vec![0u8; 524_288]; for (index, chunk_path) in &chunks_meta { @@ -672,7 +676,11 @@ impl ChunkedUploadService { if n == 0 { break; } - hasher.update(&buf[..n]); + if use_rayon { + hasher.update_rayon(&buf[..n]); + } else { + hasher.update(&buf[..n]); + } output.write_all(&buf[..n]).map_err(|e| { format!("Failed to write chunk {index} to assembled file: {e}") })?; diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 8e9f887a..aa273b7d 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -51,6 +51,10 @@ use crate::domain::errors::{DomainError, ErrorKind}; /// Block size for BLAKE3 file hashing (1MB — optimal syscall/throughput ratio). const HASH_BLOCK_SIZE: usize = 1024 * 1024; +/// Files larger than this threshold use multithreaded BLAKE3 hashing via +/// `update_rayon()`, which splits the work across all available cores. +const RAYON_HASH_THRESHOLD: u64 = 10 * 1024 * 1024; // 10 MB + /// Chunk size for streaming file reads (256 KB) const STREAM_CHUNK_SIZE: usize = 256 * 1024; @@ -155,30 +159,49 @@ impl DedupService { // ── Hash helpers ───────────────────────────────────────────── /// Calculate BLAKE3 hash of content (~5× faster than SHA-256). + /// + /// For buffers larger than 10 MB the computation is parallelised across + /// all available cores via `update_rayon()`. pub fn hash_bytes(content: &[u8]) -> String { - blake3::hash(content).to_hex().to_string() + if content.len() as u64 > RAYON_HASH_THRESHOLD { + let mut hasher = blake3::Hasher::new(); + hasher.update_rayon(content); + hasher.finalize().to_hex().to_string() + } else { + blake3::hash(content).to_hex().to_string() + } } /// Calculate BLAKE3 hash of a file (~5× faster than SHA-256). /// /// Runs entirely on `spawn_blocking` with synchronous I/O so the Tokio - /// worker threads are never blocked by CPU-bound hashing. Uses 1 MB - /// reads for optimal syscall-to-throughput ratio. + /// worker threads are never blocked by CPU-bound hashing. + /// + /// For files larger than 10 MB the hash is computed with `update_rayon()`, + /// which splits the work across all available cores. Smaller files use + /// sequential 1 MB reads for optimal syscall-to-throughput ratio. pub async fn hash_file(path: &Path) -> std::io::Result { let path = path.to_path_buf(); tokio::task::spawn_blocking(move || { - use std::io::Read; - - let mut file = std::fs::File::open(&path)?; + let file_size = std::fs::metadata(&path)?.len(); let mut hasher = blake3::Hasher::new(); - let mut buffer = vec![0u8; HASH_BLOCK_SIZE]; - loop { - let n = file.read(&mut buffer)?; - if n == 0 { - break; + if file_size > RAYON_HASH_THRESHOLD { + // Large file: read into memory and hash with all cores + let content = std::fs::read(&path)?; + hasher.update_rayon(&content); + } else { + // Small file: sequential streaming with 1 MB reads + use std::io::Read; + let mut file = std::fs::File::open(&path)?; + let mut buffer = vec![0u8; HASH_BLOCK_SIZE]; + loop { + let n = file.read(&mut buffer)?; + if n == 0 { + break; + } + hasher.update(&buffer[..n]); } - hasher.update(&buffer[..n]); } Ok(hasher.finalize().to_hex().to_string()) diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 37c3dde9..fbaef4dd 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -12,5 +12,6 @@ pub mod path_resolver_service; pub mod path_service; pub mod thumbnail_service; pub mod trash_cleanup_service; +pub mod webdav_lock_service; pub mod wopi_discovery_service; pub mod zip_service; diff --git a/src/infrastructure/services/webdav_lock_service.rs b/src/infrastructure/services/webdav_lock_service.rs new file mode 100644 index 00000000..bde0ebc9 --- /dev/null +++ b/src/infrastructure/services/webdav_lock_service.rs @@ -0,0 +1,218 @@ +//! WebDAV lock store backed by Moka (in-memory cache with per-entry TTL). +//! +//! Locks are automatically evicted when their timeout expires, preventing +//! orphaned locks from accumulating. Two caches are maintained: +//! +//! - `by_path` : path → `LockEntry` (for LOCK conflict detection) +//! - `by_token` : token → path (for fast UNLOCK / refresh lookups) +//! +//! Both caches share the same TTL so entries disappear together. + +use std::sync::Arc; +use std::time::Duration; + +use crate::application::adapters::webdav_adapter::{LockInfo, LockScope}; + +/// Default lock timeout when the client does not specify one (RFC 4918 §10.7). +const DEFAULT_LOCK_TIMEOUT_SECS: u64 = 1800; // 30 minutes + +/// Absolute maximum TTL a client may request. +const MAX_LOCK_TIMEOUT_SECS: u64 = 86_400; // 24 hours + +/// A stored lock entry. +#[derive(Clone, Debug)] +pub struct LockEntry { + pub info: LockInfo, + pub path: String, +} + +/// In-memory WebDAV lock store with automatic TTL-based expiration. +/// +/// Uses Moka's `sync::Cache` — lock-free (sharded) reads, bounded size, +/// and per-entry TTL via `policy::Expiry`. +pub struct WebDavLockStore { + /// path → `LockEntry` + by_path: moka::sync::Cache, + /// token → path (reverse index) + by_token: moka::sync::Cache, +} + +impl WebDavLockStore { + /// Create a new lock store. + /// + /// * `max_capacity` — upper bound on simultaneous locks (evicts LRU on overflow). + pub fn new(max_capacity: u64) -> Self { + // We use `expire_after` (per-entry TTL) via insert with explicit ttl, + // so we configure a generous global time_to_live as a safety net. + let by_path = moka::sync::Cache::builder() + .max_capacity(max_capacity) + .time_to_live(Duration::from_secs(MAX_LOCK_TIMEOUT_SECS)) + .build(); + + let by_token = moka::sync::Cache::builder() + .max_capacity(max_capacity) + .time_to_live(Duration::from_secs(MAX_LOCK_TIMEOUT_SECS)) + .build(); + + Self { by_path, by_token } + } + + // ── Public API ────────────────────────────────────────────── + + /// Attempt to acquire a lock on `path`. + /// + /// Returns `Ok(LockEntry)` on success, or `Err(existing)` if the resource + /// is already exclusively locked by a different token. + pub fn acquire( + &self, + path: &str, + info: LockInfo, + ) -> Result { + // Check for existing conflicting lock + if let Some(existing) = self.by_path.get(path) { + if existing.info.scope == LockScope::Exclusive { + return Err(existing); + } + } + + let ttl = Self::parse_timeout(info.timeout.as_deref()); + let entry = LockEntry { + info, + path: path.to_owned(), + }; + + self.by_path + .insert(path.to_owned(), entry.clone()); + self.by_token + .insert(entry.info.token.clone(), path.to_owned()); + + // Moka 0.12 does not expose per-entry set_expiration_after_insert at + // insert time. We rely on the global `time_to_live` as an upper bound + // and use the `invalidate_after` helper below for custom TTL. + // + // To implement shorter-than-max TTL we schedule an async invalidation. + if ttl.as_secs() < MAX_LOCK_TIMEOUT_SECS { + let by_path = self.by_path.clone(); + let by_token = self.by_token.clone(); + let token = entry.info.token.clone(); + let path_owned = path.to_owned(); + tokio::spawn(async move { + tokio::time::sleep(ttl).await; + // Only remove if the entry still matches (wasn't refreshed/replaced) + if let Some(e) = by_path.get(&path_owned) { + if e.info.token == token { + by_path.invalidate(&path_owned); + by_token.invalidate(&token); + } + } + }); + } + + Ok(entry) + } + + /// Refresh an existing lock (extend its timeout). + /// + /// Returns `Some(LockEntry)` with updated timeout, or `None` if the token + /// is unknown (expired or never existed). + pub fn refresh(&self, token: &str, new_timeout: Option<&str>) -> Option { + let path = self.by_token.get(token)?; + let mut entry = self.by_path.get(&path)?; + + if entry.info.token != token { + return None; // token mismatch — lock was replaced + } + + let ttl = Self::parse_timeout(new_timeout.or(entry.info.timeout.as_deref())); + let timeout_str = format!("Second-{}", ttl.as_secs()); + entry.info.timeout = Some(timeout_str.clone()); + + // Re-insert to reset the TTL + self.by_path.insert(path.clone(), entry.clone()); + self.by_token.insert(token.to_owned(), path.clone()); + + if ttl.as_secs() < MAX_LOCK_TIMEOUT_SECS { + let by_path = self.by_path.clone(); + let by_token = self.by_token.clone(); + let token_owned = token.to_owned(); + let path_owned = path.clone(); + tokio::spawn(async move { + tokio::time::sleep(ttl).await; + if let Some(e) = by_path.get(&path_owned) { + if e.info.token == token_owned { + by_path.invalidate(&path_owned); + by_token.invalidate(&token_owned); + } + } + }); + } + + Some(entry) + } + + /// Release a lock by its token. + /// + /// Returns `true` if the lock existed and was removed. + pub fn release(&self, token: &str) -> bool { + if let Some(path) = self.by_token.get(token) { + // Only remove from by_path if the token still matches + if let Some(entry) = self.by_path.get(&path) { + if entry.info.token == token { + self.by_path.invalidate(&path); + } + } + self.by_token.invalidate(token); + true + } else { + false + } + } + + /// Look up a lock by resource path. + pub fn get_by_path(&self, path: &str) -> Option { + self.by_path.get(path) + } + + /// Look up a lock by token. + pub fn get_by_token(&self, token: &str) -> Option { + let path = self.by_token.get(token)?; + self.by_path.get(&path) + } + + // ── Helpers ───────────────────────────────────────────────── + + /// Parse a WebDAV `Timeout` header value into a [`Duration`]. + /// + /// Accepted formats (RFC 4918 §10.7): + /// - `Second-NNN` + /// - `Infinite` (clamped to `MAX_LOCK_TIMEOUT_SECS`) + /// - Comma-separated list (first value wins) + fn parse_timeout(header: Option<&str>) -> Duration { + let raw = match header { + Some(v) if !v.is_empty() => v, + _ => return Duration::from_secs(DEFAULT_LOCK_TIMEOUT_SECS), + }; + + // Take the first value in a comma-separated list + let first = raw.split(',').next().unwrap_or(raw).trim(); + + if first.eq_ignore_ascii_case("Infinite") { + return Duration::from_secs(MAX_LOCK_TIMEOUT_SECS); + } + + if let Some(secs_str) = first.strip_prefix("Second-") { + if let Ok(secs) = secs_str.trim().parse::() { + return Duration::from_secs(secs.min(MAX_LOCK_TIMEOUT_SECS)); + } + } + + Duration::from_secs(DEFAULT_LOCK_TIMEOUT_SECS) + } +} + +/// Create a shared lock store wrapped in `Arc` for embedding in `AppState`. +pub fn create_webdav_lock_store() -> Arc { + // 10 000 simultaneous locks should be more than enough; Moka evicts LRU + // if the cap is reached, so stale entries are cleaned automatically. + Arc::new(WebDavLockStore::new(10_000)) +} diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index 03620572..0c13212e 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -18,7 +18,7 @@ use quick_xml::Writer; use uuid::Uuid; use crate::application::adapters::webdav_adapter::{ - LockInfo, LockScope, LockType, PropFindRequest, WebDavAdapter, + LockInfo, PropFindRequest, WebDavAdapter, }; use crate::application::dtos::file_dto::FileDto; use crate::application::dtos::folder_dto::FolderDto; @@ -1481,7 +1481,7 @@ async fn handle_copy( * @return XML response with lock information */ async fn handle_lock( - _state: Arc, + state: Arc, req: Request, path: String, ) -> Result, AppError> { @@ -1524,9 +1524,10 @@ async fn handle_lock( .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))? }; + let lock_store = &state.webdav_lock_store; + // Check if this is a lock refresh (If header with a lock token) if let Some(if_header) = if_header_value { - // This is a lock refresh request // Extract lock token from If header let token = if_header .trim() @@ -1534,30 +1535,24 @@ async fn handle_lock( .trim_end_matches(">)") .to_string(); - // In a full implementation, we would look up the lock in a database - // and refresh its timeout. For now, just respond as if we did. - - // Generate lock token and owner (for a real implementation, we'd store these) - let lock_info = LockInfo { - token, - owner: Some(user.id.clone()), - depth: depth.to_string(), - timeout, - scope: LockScope::Exclusive, // Default to exclusive - type_: LockType::Write, // Default to write - }; + // Refresh the lock in the store (extends TTL) + let entry = lock_store + .refresh(&token, timeout.as_deref()) + .ok_or_else(|| { + AppError::precondition_failed(format!("Lock token not found or expired: {}", token)) + })?; // Generate response let href = format!("/webdav/{}", encode_uri_path(&path)); let mut response_body = Vec::new(); - WebDavAdapter::generate_lock_response(&mut response_body, &lock_info, &href).map_err( + WebDavAdapter::generate_lock_response(&mut response_body, &entry.info, &href).map_err( |e| AppError::internal_error(format!("Failed to generate LOCK response: {}", e)), )?; Ok(Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/xml; charset=utf-8") - .header(HEADER_LOCK_TOKEN, format!("<{}>", lock_info.token)) + .header(HEADER_LOCK_TOKEN, format!("<{}>", entry.info.token)) .body(Body::from(response_body)) .unwrap()) } else if !body_bytes.is_empty() { @@ -1565,7 +1560,6 @@ async fn handle_lock( let (scope, type_, owner) = WebDavAdapter::parse_lockinfo(body_bytes.reader()) .map_err(|e| AppError::bad_request(format!("Failed to parse LOCK request: {}", e)))?; - // Generate lock token and owner (for a real implementation, we'd store these) let token = format!("opaquelocktoken:{}", Uuid::new_v4()); let lock_info = LockInfo { token, @@ -1576,17 +1570,25 @@ async fn handle_lock( type_, }; + // Try to acquire the lock (conflict detection via moka store) + let entry = lock_store.acquire(&path, lock_info).map_err(|existing| { + AppError::locked(format!( + "Resource already locked by token {}", + existing.info.token + )) + })?; + // Generate response let href = format!("/webdav/{}", encode_uri_path(&path)); let mut response_body = Vec::new(); - WebDavAdapter::generate_lock_response(&mut response_body, &lock_info, &href).map_err( + WebDavAdapter::generate_lock_response(&mut response_body, &entry.info, &href).map_err( |e| AppError::internal_error(format!("Failed to generate LOCK response: {}", e)), )?; Ok(Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/xml; charset=utf-8") - .header(HEADER_LOCK_TOKEN, format!("<{}>", lock_info.token)) + .header(HEADER_LOCK_TOKEN, format!("<{}>", entry.info.token)) .body(Body::from(response_body)) .unwrap()) } else { @@ -1607,7 +1609,7 @@ async fn handle_lock( * @return HTTP response indicating success */ async fn handle_unlock( - _state: Arc, + state: Arc, req: Request, _path: String, ) -> Result, AppError> { @@ -1627,14 +1629,20 @@ async fn handle_unlock( .ok_or_else(|| AppError::bad_request("Lock-Token header required"))?; // Extract token from header value (format: ) - let _token = lock_token + let token = lock_token .trim() .trim_start_matches('<') .trim_end_matches('>') .to_string(); - // In a full implementation, we would look up the lock in a database - // and remove it. For now, just respond as if we did. + // Remove the lock from the store + if !state.webdav_lock_store.release(&token) { + // RFC 4918 §9.11.1: If the lock does not exist, return 409 Conflict + return Err(AppError::conflict(format!( + "Lock token not found or already expired: {}", + token + ))); + } Ok(Response::builder() .status(StatusCode::NO_CONTENT) diff --git a/src/interfaces/errors.rs b/src/interfaces/errors.rs index 47b9053a..9e55ef92 100644 --- a/src/interfaces/errors.rs +++ b/src/interfaces/errors.rs @@ -78,6 +78,11 @@ impl AppError { Self::new(StatusCode::CONFLICT, message, "Conflict") } + /// Create a 423 Locked error (WebDAV). + pub fn locked(message: impl Into) -> Self { + Self::new(StatusCode::LOCKED, message, "Locked") + } + /// Create a 415 Unsupported Media Type error. pub fn unsupported_media_type(message: impl Into) -> Self { Self::new(