merge: resolve conflict in auth_factory.rs with origin/main
Keep custom Result type alias (replacing anyhow) from main while preserving our removal of unused TokenServicePort import. https://claude.ai/code/session_01EbAFEfyJNLRmJHmmYDX3Tt
This commit is contained in:
@@ -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}")
|
||||
})?;
|
||||
|
||||
@@ -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<String> {
|
||||
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())
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String, LockEntry>,
|
||||
/// token → path (reverse index)
|
||||
by_token: moka::sync::Cache<String, String>,
|
||||
}
|
||||
|
||||
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<LockEntry, LockEntry> {
|
||||
// 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<LockEntry> {
|
||||
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<LockEntry> {
|
||||
self.by_path.get(path)
|
||||
}
|
||||
|
||||
/// Look up a lock by token.
|
||||
pub fn get_by_token(&self, token: &str) -> Option<LockEntry> {
|
||||
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::<u64>() {
|
||||
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<WebDavLockStore> {
|
||||
// 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))
|
||||
}
|
||||
Reference in New Issue
Block a user