Merge pull request #440 from AtalayaLabs/claude/tender-shannon-b65gyk

src/domain/entities/file.rs

@@ -388,45 +388,36 @@ impl File {
     // Methods to create new versions of the file (immutable)
 
     /// Creates a new version of the file with updated name
-    pub fn with_name(&self, new_name: String) -> FileResult<Self> {
+    pub fn with_name(mut self, new_name: String) -> FileResult<Self> {
         let new_name = normalize_storage_name(&new_name);
         if let Err(reason) = validate_storage_name(&new_name) {
             return Err(FileError::InvalidFileName(format!("{new_name}: {reason}")));
         }
 
-        // Update path based on name
+        // Recompute the path from the unchanged parent + the new name.
-        let parent_path = self.storage_path.parent();
+        let new_storage_path = match self.storage_path.parent() {
-        let new_storage_path = match parent_path {
             Some(parent) => parent.join(&new_name),
             None => StoragePath::from_string(&new_name),
         };
 
-        // Update string representation
-        let new_path_string = new_storage_path.to_string();
-
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
             .as_secs();
 
-        Ok(Self {
+        // Consume `self` and mutate in place — only the path, name and mtime
-            id: self.id.clone(),
+        // change; id / mime_type / folder_id / blob_hash are carried over
-            name: new_name,
+        // without the per-field clone the old `&self` builder paid.
-            storage_path: new_storage_path,
+        self.path_string = new_storage_path.to_string();
-            path_string: new_path_string,
+        self.storage_path = new_storage_path;
-            size: self.size,
+        self.name = new_name;
-            mime_type: self.mime_type.clone(),
+        self.modified_at = now;
-            folder_id: self.folder_id.clone(),
+        Ok(self)
-            created_at: self.created_at,
-            modified_at: now,
-            owner_id: self.owner_id,
-            blob_hash: self.blob_hash.clone(),
-        })
     }
 
     /// Creates a new version of the file with updated folder
     pub fn with_folder(
-        &self,
+        mut self,
         folder_id: Option<String>,
         folder_path: Option<StoragePath>,
     ) -> FileResult<Self> {
@@ -436,49 +427,30 @@ impl File {
             None => StoragePath::from_string(&self.name), // Root
         };
 
-        // Update string representation
-        let new_path_string = new_storage_path.to_string();
-
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
             .as_secs();
 
-        Ok(Self {
+        // Consume `self`: only the path, folder_id and mtime change.
-            id: self.id.clone(),
+        self.path_string = new_storage_path.to_string();
-            name: self.name.clone(),
+        self.storage_path = new_storage_path;
-            storage_path: new_storage_path,
+        self.folder_id = folder_id;
-            path_string: new_path_string,
+        self.modified_at = now;
-            size: self.size,
+        Ok(self)
-            mime_type: self.mime_type.clone(),
-            folder_id,
-            created_at: self.created_at,
-            modified_at: now,
-            owner_id: self.owner_id,
-            blob_hash: self.blob_hash.clone(),
-        })
     }
 
     /// Creates a new version of the file with updated size
-    pub fn with_size(&self, new_size: u64) -> Self {
+    pub fn with_size(mut self, new_size: u64) -> Self {
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
             .as_secs();
 
-        Self {
+        // Consume `self`: only size and mtime change — no per-field clone.
-            id: self.id.clone(),
+        self.size = new_size;
-            name: self.name.clone(),
+        self.modified_at = now;
-            storage_path: self.storage_path.clone(),
+        self
-            path_string: self.path_string.clone(),
-            size: new_size,
-            mime_type: self.mime_type.clone(),
-            folder_id: self.folder_id.clone(),
-            created_at: self.created_at,
-            modified_at: now,
-            owner_id: self.owner_id,
-            blob_hash: self.blob_hash.clone(),
-        }
     }
 }
 

src/domain/services/path_service.rs

@@ -103,18 +103,16 @@ impl StoragePath {
         Self { segments }
     }
 
-    /// Appends a segment to the path.
+    /// Appends a segment to the path, consuming `self` so the existing
+    /// segment buffer is reused instead of deep-cloned.
     ///
     /// Traversal segments (`.`, `..`) and segments containing `/` are
     /// silently ignored to prevent path-traversal attacks.
-    pub fn join(&self, segment: &str) -> Self {
+    pub fn join(mut self, segment: &str) -> Self {
-        let mut new_segments = self.segments.clone();
         if Self::is_safe_segment(segment) {
-            new_segments.push(segment.to_string());
+            self.segments.push(segment.to_string());
-        }
-        Self {
-            segments: new_segments,
         }
+        self
     }
 
     /// Gets the file name (last segment)

src/infrastructure/db.rs

@@ -114,6 +114,12 @@ async fn create_pool_with_retries(
             .acquire_timeout(Duration::from_secs(connect_timeout_secs))
             .idle_timeout(Duration::from_secs(idle_timeout_secs))
             .max_lifetime(Duration::from_secs(max_lifetime_secs))
+            // Skip the liveness ping sqlx issues on every acquire() (on by
+            // default): with warm min_connections and a bounded max_lifetime,
+            // that extra round-trip per checkout costs more than the rare dead
+            // connection it catches. A stale socket surfaces as a query error
+            // and the pool recycles it either way.
+            .test_before_acquire(false)
             .connect(connection_string)
             .await
         {

src/infrastructure/repositories/pg/subject_group_pg_repository.rs

@@ -126,30 +126,26 @@ impl SubjectGroupRepository for SubjectGroupPgRepository {
         offset: u32,
         name_query: Option<&str>,
     ) -> Result<(Vec<SubjectGroup>, u64), SubjectGroupRepositoryError> {
-        // Two queries: one for the page, one for the total count. The query
+        // Single query: the page plus `COUNT(*) OVER()` for the total matching
-        // is small and frequent; a window function would add complexity for
+        // count, folding what used to be a separate COUNT round-trip into one.
-        // no measurable win.
+        let (sql_page, pattern) = match name_query {
-        let (sql_page, sql_count, pattern) = match name_query {
+            Some(q) => (
-            Some(q) => {
+                "SELECT id, name, description, is_virtual, created_at, updated_at,
-                let pat = like_escape(q);
+                        COUNT(*) OVER() AS total_count
-                (
-                    "SELECT id, name, description, is_virtual, created_at, updated_at
                  FROM auth.subject_groups
                  WHERE name ILIKE $1
                  ORDER BY is_virtual DESC, name
                  LIMIT $2 OFFSET $3"
                     .to_string(),
-                    "SELECT COUNT(*) FROM auth.subject_groups WHERE name ILIKE $1".to_string(),
+                Some(like_escape(q)),
-                    Some(pat),
+            ),
-                )
-            }
             None => (
-                "SELECT id, name, description, is_virtual, created_at, updated_at
+                "SELECT id, name, description, is_virtual, created_at, updated_at,
+                        COUNT(*) OVER() AS total_count
                  FROM auth.subject_groups
                  ORDER BY is_virtual DESC, name
                  LIMIT $1 OFFSET $2"
                     .to_string(),
-                "SELECT COUNT(*) FROM auth.subject_groups".to_string(),
                 None,
             ),
         };
@@ -170,19 +166,9 @@ impl SubjectGroupRepository for SubjectGroupPgRepository {
         }
         .map_err(|e| Self::map_sqlx_err("list page", e))?;
 
-        let total: i64 = if let Some(ref p) = pattern {
+        // total_count is identical in every row; 0 when the page is empty.
-            sqlx::query_scalar(&sql_count)
+        let total = rows.first().map_or(0, |r| r.get::<i64, _>("total_count")) as u64;
-                .bind(p)
+        Ok((rows.iter().map(Self::row_to_group).collect(), total))
-                .fetch_one(self.pool.as_ref())
-                .await
-        } else {
-            sqlx::query_scalar(&sql_count)
-                .fetch_one(self.pool.as_ref())
-                .await
-        }
-        .map_err(|e| Self::map_sqlx_err("list count", e))?;
-
-        Ok((rows.iter().map(Self::row_to_group).collect(), total as u64))
     }
 
     async fn list_with_counts(
@@ -191,16 +177,15 @@ impl SubjectGroupRepository for SubjectGroupPgRepository {
         offset: u32,
         name_query: Option<&str>,
     ) -> Result<(Vec<(SubjectGroup, i64)>, u64), SubjectGroupRepositoryError> {
-        // Single SQL: groups + COUNT of direct members per group, via LEFT JOIN
+        // Single SQL: groups + per-group member COUNT via LEFT JOIN on
-        // on `auth.subject_group_members`. No N+1; one round-trip for the
+        // `auth.subject_group_members`, plus `COUNT(*) OVER()` for the total
-        // page, a second for the unfiltered total (matches `list`).
+        // group count. No N+1 and no separate COUNT round-trip — one query.
-        let (sql_page, sql_count, pattern) = match name_query {
+        let (sql_page, pattern) = match name_query {
-            Some(q) => {
+            Some(q) => (
-                let pat = like_escape(q);
-                (
                 "SELECT g.id, g.name, g.description, g.is_virtual,
                         g.created_at, g.updated_at,
-                            COUNT(m.group_id) AS member_count
+                        COUNT(m.group_id) AS member_count,
+                        COUNT(*) OVER() AS total_count
                  FROM auth.subject_groups g
                  LEFT JOIN auth.subject_group_members m ON m.group_id = g.id
                  WHERE g.name ILIKE $1
@@ -208,21 +193,19 @@ impl SubjectGroupRepository for SubjectGroupPgRepository {
                  ORDER BY g.is_virtual DESC, g.name
                  LIMIT $2 OFFSET $3"
                     .to_string(),
-                    "SELECT COUNT(*) FROM auth.subject_groups WHERE name ILIKE $1".to_string(),
+                Some(like_escape(q)),
-                    Some(pat),
+            ),
-                )
-            }
             None => (
                 "SELECT g.id, g.name, g.description, g.is_virtual,
                         g.created_at, g.updated_at,
-                        COUNT(m.group_id) AS member_count
+                        COUNT(m.group_id) AS member_count,
+                        COUNT(*) OVER() AS total_count
                  FROM auth.subject_groups g
                  LEFT JOIN auth.subject_group_members m ON m.group_id = g.id
                  GROUP BY g.id
                  ORDER BY g.is_virtual DESC, g.name
                  LIMIT $1 OFFSET $2"
                     .to_string(),
-                "SELECT COUNT(*) FROM auth.subject_groups".to_string(),
                 None,
             ),
         };
@@ -243,24 +226,14 @@ impl SubjectGroupRepository for SubjectGroupPgRepository {
         }
         .map_err(|e| Self::map_sqlx_err("list_with_counts page", e))?;
 
-        let total: i64 = if let Some(ref p) = pattern {
+        // total_count is identical in every row; 0 when the page is empty.
-            sqlx::query_scalar(&sql_count)
+        let total = rows.first().map_or(0, |r| r.get::<i64, _>("total_count")) as u64;
-                .bind(p)
-                .fetch_one(self.pool.as_ref())
-                .await
-        } else {
-            sqlx::query_scalar(&sql_count)
-                .fetch_one(self.pool.as_ref())
-                .await
-        }
-        .map_err(|e| Self::map_sqlx_err("list_with_counts total", e))?;
-
         let items = rows
             .iter()
             .map(|r| (Self::row_to_group(r), r.get::<i64, _>("member_count")))
             .collect();
 
-        Ok((items, total as u64))
+        Ok((items, total))
     }
 
     async fn count_members(&self, id: Uuid) -> Result<i64, SubjectGroupRepositoryError> {

src/infrastructure/services/cached_blob_backend.rs

@@ -96,9 +96,11 @@ impl BlobStorageBackend for CachedBlobBackend {
                 DomainError::internal_error("BlobCache", format!("mkdir cache_dir: {e}"))
             })?;
 
-            // Scan existing cache to rebuild index
+            // Scan existing cache to rebuild index.  Collect entries WITHOUT
+            // holding the index lock — a large cache directory walk must not
+            // serialize concurrent blob operations behind the mutex.
             let mut total_bytes = 0u64;
-            let mut idx = index.lock().await;
+            let mut entries: Vec<(String, u64)> = Vec::new();
             if let Ok(mut read_dir) = fs::read_dir(&cache_dir).await {
                 while let Ok(Some(prefix_entry)) = read_dir.next_entry().await {
                     if !prefix_entry.path().is_dir() {
@@ -111,14 +113,20 @@ impl BlobStorageBackend for CachedBlobBackend {
                                 && let Some(stem) = path.file_stem().and_then(|s| s.to_str())
                             {
                                 let size = fs::metadata(&path).await.map(|m| m.len()).unwrap_or(0);
-                                idx.put(stem.to_string(), CacheEntry { size });
+                                entries.push((stem.to_string(), size));
                                 total_bytes += size;
                             }
                         }
                     }
                 }
             }
-            drop(idx);
+            // Bulk-insert the rebuilt index under a single brief lock.
+            {
+                let mut idx = index.lock().await;
+                for (stem, size) in entries {
+                    idx.put(stem, CacheEntry { size });
+                }
+            }
             current_size.store(total_bytes, Ordering::Relaxed);
             tracing::info!(
                 "Blob cache initialized: {} bytes in cache at {}",
@@ -196,21 +204,20 @@ impl BlobStorageBackend for CachedBlobBackend {
         let max_cache_bytes = self.max_cache_bytes;
         let current_size = self.current_size.clone();
         Box::pin(async move {
-            // Check cache
+            // Check cache presence (and bump LRU recency) under a brief lock,
-            {
+            // then release it BEFORE touching the filesystem so concurrent
-                let mut idx = index.lock().await;
+            // readers don't serialize behind a single open() syscall.
-                if idx.get(&hash).is_some() {
+            if index.lock().await.get(&hash).is_some() {
                 if let Ok(file) = fs::File::open(&cached).await {
                     let stream: BlobStream =
                         Box::pin(ReaderStream::with_capacity(file, STREAM_CHUNK_SIZE));
                     return Ok(stream);
                 }
-                    // Cache entry stale — remove
+                // Cache entry stale (file vanished) — drop it from the index.
-                    if let Some(entry) = idx.pop(&hash) {
+                if let Some(entry) = index.lock().await.pop(&hash) {
                     current_size.fetch_sub(entry.size, Ordering::Relaxed);
                 }
             }
-            }
 
             // Cache miss — fetch from inner, spool to cache
             let self_ref = CachedRef {
@@ -243,10 +250,10 @@ impl BlobStorageBackend for CachedBlobBackend {
         let max_cache_bytes = self.max_cache_bytes;
         let current_size = self.current_size.clone();
         Box::pin(async move {
-            // Try cache first
+            // Check cache presence (and bump LRU recency) under a brief lock,
-            {
+            // then release it BEFORE the open()/seek() syscalls so concurrent
-                let mut idx = index.lock().await;
+            // range readers don't serialize behind the index mutex.
-                if idx.get(&hash).is_some() {
+            if index.lock().await.get(&hash).is_some() {
                 if let Ok(mut file) = fs::File::open(&cached).await {
                     file.seek(std::io::SeekFrom::Start(start))
                         .await
@@ -259,11 +266,10 @@ impl BlobStorageBackend for CachedBlobBackend {
                         Box::pin(ReaderStream::with_capacity(limited, STREAM_CHUNK_SIZE));
                     return Ok(stream);
                 }
-                    if let Some(entry) = idx.pop(&hash) {
+                if let Some(entry) = index.lock().await.pop(&hash) {
                     current_size.fetch_sub(entry.size, Ordering::Relaxed);
                 }
             }
-            }
 
             // Cache miss — fetch full blob into cache, then serve range
             let self_ref = CachedRef {
@@ -298,9 +304,9 @@ impl BlobStorageBackend for CachedBlobBackend {
         let current_size = self.current_size.clone();
         Box::pin(async move {
             inner.delete_blob(&hash).await?;
-            // Remove from cache
+            // Remove from cache — drop the index lock before the unlink()
-            let mut idx = index.lock().await;
+            // syscall so deletes don't serialize concurrent cache lookups.
-            if let Some(entry) = idx.pop(&hash) {
+            if let Some(entry) = index.lock().await.pop(&hash) {
                 current_size.fetch_sub(entry.size, Ordering::Relaxed);
             }
             let _ = fs::remove_file(&cached).await;
@@ -402,6 +408,26 @@ impl CachedRef {
         self.cache_dir.join(prefix).join(format!("{hash}.blob"))
     }
 
+    /// Pop LRU entries until the cache is back within its byte budget,
+    /// returning the on-disk paths of the evicted blobs.
+    ///
+    /// Only the in-memory index is touched here (atomic counter + LRU map);
+    /// the caller MUST unlink the returned paths AFTER releasing the index
+    /// lock so the `remove_file` syscalls never run while the mutex is held.
+    fn collect_evictions(&self, idx: &mut LruCache<String, CacheEntry>) -> Vec<PathBuf> {
+        let mut victims = Vec::new();
+        while self.current_size.load(Ordering::Relaxed) > self.max_cache_bytes {
+            if let Some((evicted_hash, evicted_entry)) = idx.pop_lru() {
+                self.current_size
+                    .fetch_sub(evicted_entry.size, Ordering::Relaxed);
+                victims.push(self.cached_path(&evicted_hash));
+            } else {
+                break;
+            }
+        }
+        victims
+    }
+
     async fn insert_into_cache_static(
         &self,
         hash: &str,
@@ -423,21 +449,19 @@ impl CachedRef {
             DomainError::internal_error("BlobCache", format!("cache copy failed: {e}"))
         })?;
 
+        // Update the index and pick eviction victims under a single brief
+        // lock, then unlink the evicted files AFTER releasing it — file
+        // removal must not run while the index mutex is held.
+        let to_evict = {
             let mut idx = self.index.lock().await;
             if let Some(old) = idx.put(hash.to_string(), CacheEntry { size }) {
                 self.current_size.fetch_sub(old.size, Ordering::Relaxed);
             }
             self.current_size.fetch_add(size, Ordering::Relaxed);
-
+            self.collect_evictions(&mut idx)
-        while self.current_size.load(Ordering::Relaxed) > self.max_cache_bytes {
+        };
-            if let Some((evicted_hash, evicted_entry)) = idx.pop_lru() {
+        for path in to_evict {
-                self.current_size
+            let _ = fs::remove_file(&path).await;
-                    .fetch_sub(evicted_entry.size, Ordering::Relaxed);
-                let evicted_path = self.cached_path(&evicted_hash);
-                let _ = fs::remove_file(&evicted_path).await;
-            } else {
-                break;
-            }
         }
         Ok(())
     }
@@ -482,21 +506,16 @@ impl CachedRef {
             .await
             .map_err(|e| DomainError::internal_error("BlobCache", format!("rename: {e}")))?;
 
+        let to_evict = {
             let mut idx = self.index.lock().await;
             if let Some(old) = idx.put(hash.to_string(), CacheEntry { size: total }) {
                 self.current_size.fetch_sub(old.size, Ordering::Relaxed);
             }
             self.current_size.fetch_add(total, Ordering::Relaxed);
-
+            self.collect_evictions(&mut idx)
-        while self.current_size.load(Ordering::Relaxed) > self.max_cache_bytes {
+        };
-            if let Some((evicted_hash, evicted_entry)) = idx.pop_lru() {
+        for path in to_evict {
-                self.current_size
+            let _ = fs::remove_file(&path).await;
-                    .fetch_sub(evicted_entry.size, Ordering::Relaxed);
-                let evicted_path = self.cached_path(&evicted_hash);
-                let _ = fs::remove_file(&evicted_path).await;
-            } else {
-                break;
-            }
         }
 
         Ok(dest)

src/infrastructure/services/dedup_service.rs

@@ -459,8 +459,12 @@ impl DedupService {
     /// exist in `storage.blobs`.
     /// Phase 1: Reads only *new* chunks from the source file (the biggest
     /// I/O saving for versioned files where most chunks are unchanged).
-    /// Phase 2: Parallel operations — uploads new chunks, bumps ref_count
+    /// Uploads each new chunk and bumps `ref_count` for chunks that already
-    /// for existing ones — with up to [`CHUNK_UPLOAD_CONCURRENCY`] in flight.
+    /// exist, with up to [`CHUNK_UPLOAD_CONCURRENCY`] uploads in flight.
+    ///
+    /// `ref_count` is incremented once per *distinct* chunk (one reference per
+    /// manifest), staying symmetric with `remove_manifest_reference` so a file
+    /// that repeats a chunk cannot over-count and leak the blob forever.
     async fn store_chunks(
         &self,
         source_path: &Path,
@@ -469,20 +473,22 @@ impl DedupService {
         let pool = &self.pool;
         let backend = &self.backend;
 
-        // ── Phase 0: Batch-check which chunks already exist ──────
+        // ── Phase 0: de-duplicate chunk hashes, then batch-check existence ──
-        let unique_hashes: Vec<String> = {
+        // A single file can legitimately repeat the same chunk many times
+        // (zero-filled regions in disk/VM images, repeated document structures,
+        // concatenated archives). `ref_count` is tracked per *distinct* chunk —
+        // one reference per manifest — to stay symmetric with
+        // `remove_manifest_reference`, which decrements via
+        // `WHERE hash = ANY(chunk_hashes)` (matching each row once). Counting
+        // per-occurrence here would over-increment and leak the blob forever.
+        // Keep the first occurrence of each hash so new chunks know where to
+        // read their bytes.
         let mut seen = std::collections::HashSet::new();
-            chunks
+        let unique_chunks: Vec<&ChunkMeta> = chunks
             .iter()
-                .filter_map(|c| {
+            .filter(|c| seen.insert(c.hash.as_str()))
-                    if seen.insert(c.hash.as_str()) {
+            .collect();
-                        Some(c.hash.clone())
+        let unique_hashes: Vec<String> = unique_chunks.iter().map(|c| c.hash.clone()).collect();
-                    } else {
-                        None
-                    }
-                })
-                .collect()
-        };
 
         let existing_hashes: std::collections::HashSet<String> =
             sqlx::query_scalar::<_, String>("SELECT hash FROM storage.blobs WHERE hash = ANY($1)")
@@ -498,61 +504,50 @@ impl DedupService {
                 .into_iter()
                 .collect();
 
-        // ── Phase 1+2 (fused): upload NEW chunks just-in-time ────
+        // ── Phase 1: bump ref_count for every existing chunk in ONE query ──
-        // Read each new chunk by positioned I/O immediately before its
+        // (was one UPDATE per occurrence — now a single batched round-trip).
-        // upload, instead of first materializing every new chunk's *data* in
+        let existing: Vec<String> = unique_hashes
-        // a Vec.  Peak heap for file content is bounded to
-        // ~CHUNK_UPLOAD_CONCURRENCY × CDC_MAX_CHUNK (≈ 8 MiB) — proportional
-        // to the chunk size, never the file size, so storing a large
-        // brand-new file no longer spikes RAM.  Existing chunks skip all disk
-        // I/O and just bump ref_count.
-        //
-        // We first collect *owned* per-chunk metadata (hash + offset + length
-        // + existence flag — no file data) so the stream below does not borrow
-        // the `chunks` parameter across an `.await` (which would make this
-        // future non-`Send` and break the upload handlers).
-        let chunk_ops: Vec<(String, u64, usize, bool)> = chunks
             .iter()
-            .map(|chunk| {
+            .filter(|h| existing_hashes.contains(*h))
-                let exists = existing_hashes.contains(&chunk.hash);
+            .cloned()
-                (
+            .collect();
-                    chunk.hash.clone(),
+        if !existing.is_empty() {
-                    chunk.offset as u64,
+            sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = ANY($1)")
-                    chunk.length,
+                .bind(&existing)
-                    exists,
+                .execute(pool.as_ref())
-                )
+                .await
-            })
+                .map_err(|e| {
+                    DomainError::internal_error("Dedup", format!("Failed to bump ref_count: {}", e))
+                })?;
+        }
+
+        // ── Phase 2: upload each NEW chunk once, concurrently ──────────────
+        // Read each new chunk by positioned I/O immediately before its upload
+        // instead of materializing every chunk's *data* up front. Peak heap for
+        // file content stays bounded to ~CHUNK_UPLOAD_CONCURRENCY × CDC_MAX_CHUNK
+        // (≈ 8 MiB) — proportional to the chunk size, never the file size.
+        //
+        // Owned metadata (hash + offset + length, no file data) so the stream
+        // below does not borrow `chunks` across an `.await` (which would make
+        // this future non-`Send` and break the upload handlers).
+        let new_ops: Vec<(String, u64, usize)> = unique_chunks
+            .iter()
+            .filter(|c| !existing_hashes.contains(&c.hash))
+            .map(|c| (c.hash.clone(), c.offset as u64, c.length))
             .collect();
 
         let source = Arc::new(std::fs::File::open(source_path).map_err(|e| {
             DomainError::internal_error("Dedup", format!("Failed to open source file: {}", e))
         })?);
 
-        let results: Vec<Result<(), DomainError>> = stream::iter(chunk_ops)
+        let results: Vec<Result<(), DomainError>> = stream::iter(new_ops)
-            .map(|(hash, offset, length, exists)| {
+            .map(|(hash, offset, length)| {
                 let source = source.clone();
                 let pool = pool.clone();
                 let backend = backend.clone();
                 async move {
-                    if exists {
+                    // Positioned read of just this chunk (≤ CDC_MAX_CHUNK) off
-                        // Existing chunk: bump ref_count, no disk I/O.
+                    // the async runtime, then upload.
-                        sqlx::query(
-                            "UPDATE storage.blobs
-                                SET ref_count = ref_count + 1
-                              WHERE hash = $1",
-                        )
-                        .bind(&hash)
-                        .execute(pool.as_ref())
-                        .await
-                        .map_err(|e| {
-                            DomainError::internal_error(
-                                "Dedup",
-                                format!("Failed to bump ref_count: {}", e),
-                            )
-                        })?;
-                    } else {
-                        // New chunk: positioned read of just this chunk
-                        // (≤ CDC_MAX_CHUNK) off the async runtime, then upload.
                     let bytes = tokio::task::spawn_blocking(move || {
                         use std::os::unix::fs::FileExt;
                         let mut buf = vec![0u8; length];
@@ -564,15 +559,15 @@ impl DedupService {
                         DomainError::internal_error("Dedup", format!("Read task failed: {}", e))
                     })?
                     .map_err(|e| {
-                            DomainError::internal_error(
+                        DomainError::internal_error("Dedup", format!("Failed to read chunk: {}", e))
-                                "Dedup",
-                                format!("Failed to read chunk: {}", e),
-                            )
                     })?;
 
                     backend
                         .put_blob_from_bytes(&hash, Bytes::from(bytes))
                         .await?;
+                    // ON CONFLICT covers a concurrent uploader inserting the
+                    // same brand-new chunk between the existence check above and
+                    // this INSERT.
                     sqlx::query(
                         "INSERT INTO storage.blobs (hash, size, ref_count)
                          VALUES ($1, $2, 1)
@@ -589,7 +584,6 @@ impl DedupService {
                             format!("Failed to upsert chunk: {}", e),
                         )
                     })?;
-                    }
                     Ok(())
                 }
             })
@@ -597,13 +591,12 @@ impl DedupService {
             .collect()
             .await;
 
-        // All operations must succeed.  Order preservation is not needed
-        // here — chunk_hashes/chunk_sizes are derived from the input
-        // `chunks` slice which keeps the original CDC order.
         for result in results {
             result?;
         }
 
+        // chunk_hashes/chunk_sizes keep the full per-occurrence CDC sequence —
+        // the manifest needs every chunk, in order, to reassemble the file.
         let chunk_hashes: Vec<String> = chunks.iter().map(|c| c.hash.clone()).collect();
         let chunk_sizes: Vec<u64> = chunks.iter().map(|c| c.length as u64).collect();
 

src/infrastructure/services/path_service.rs

@@ -64,7 +64,9 @@ impl PathService {
 
     /// Creates a file path within a folder
     pub fn create_file_path(&self, folder_path: &StoragePath, file_name: &str) -> StoragePath {
-        folder_path.join(file_name)
+        // `join` consumes its receiver to reuse the buffer; we only hold a
+        // borrow here, so clone first — the same copy the old `&self` join did.
+        folder_path.clone().join(file_name)
     }
 
     /// Checks if a path is a direct child of another

src/infrastructure/services/pg_acl_engine.rs

@@ -61,6 +61,18 @@ struct QueryCounters {
     expanded_groups: AtomicU32,
 }
 
+/// Defensive upper bound on the number of grant rows the *unbounded* list
+/// methods (`list_incoming_grants`, `list_grants_on_resource`) will pull into
+/// memory. These back management surfaces ("Manage sharing", "Shared with
+/// me"), not the hot `require()` path, so a single resource or subject
+/// realistically accumulates orders of magnitude fewer grants than this.
+///
+/// We fetch `MAX_GRANT_ROWS + 1` and *reject* when the cap is exceeded rather
+/// than silently truncating: `apply_role` computes an add/remove diff from the
+/// returned set, so a partial list would be acted on as if complete. Hitting
+/// the cap signals pathological data and is surfaced to operators via audit.
+const MAX_GRANT_ROWS: i64 = 10_000;
+
 pub struct PgAclEngine {
     pool: Arc<PgPool>,
     folder_repo: Arc<FolderDbRepository>,
@@ -406,6 +418,29 @@ impl PgAclEngine {
         })
     }
 
+    /// Reject an over-cap grant listing rather than returning a truncated set.
+    /// The unbounded list methods fetch `MAX_GRANT_ROWS + 1` and pass the row
+    /// count here; callers diff against the full result, so silently dropping
+    /// rows would corrupt that diff. Emits an audit line before failing so the
+    /// pathological resource/subject is visible to operators.
+    fn guard_grant_row_cap(returned: usize, op: &str) -> Result<(), DomainError> {
+        if returned as i64 > MAX_GRANT_ROWS {
+            tracing::info!(
+                target: "audit",
+                event = "authz.grant_list_rejected",
+                reason = "over_row_cap",
+                op,
+                cap = MAX_GRANT_ROWS,
+                "👮🏻‍♂️ grant listing exceeded the row safety cap; refusing to return a partial set",
+            );
+            return Err(DomainError::internal_error(
+                "PgAcl",
+                format!("{op}: too many grants (cap {})", MAX_GRANT_ROWS),
+            ));
+        }
+        Ok(())
+    }
+
     /// The actual permission decision. Wrapped by `check()` which adds
     /// per-call instrumentation.
     async fn check_inner(
@@ -525,15 +560,18 @@ impl AuthorizationEngine for PgAclEngine {
                AND subject_id   = ANY($2)
                AND ($3::text IS NULL OR permission = $3)
              ORDER BY granted_at DESC
+             LIMIT $4
             "#,
         )
         .bind(&subject_types)
         .bind(&subject_ids)
         .bind(perm_str)
+        .bind(MAX_GRANT_ROWS + 1)
         .fetch_all(self.pool.as_ref())
         .await
         .map_err(|e| DomainError::internal_error("PgAcl", format!("list incoming: {e}")))?;
 
+        Self::guard_grant_row_cap(rows.len(), "list_incoming_grants")?;
         rows.into_iter().map(Self::row_to_grant).collect()
     }
 
@@ -855,14 +893,17 @@ impl AuthorizationEngine for PgAclEngine {
              WHERE resource_type = $1
                AND resource_id   = $2
              ORDER BY granted_at DESC
+             LIMIT $3
             "#,
         )
         .bind(resource.type_str())
         .bind(resource.id())
+        .bind(MAX_GRANT_ROWS + 1)
         .fetch_all(self.pool.as_ref())
         .await
         .map_err(|e| DomainError::internal_error("PgAcl", format!("list on resource: {e}")))?;
 
+        Self::guard_grant_row_cap(rows.len(), "list_grants_on_resource")?;
         rows.into_iter().map(Self::row_to_grant).collect()
     }
 

src/interfaces/api/handlers/search_handler.rs

@@ -21,6 +21,14 @@ use std::sync::Arc;
  * formatting) is performed server-side. These handlers are thin HTTP
  * adapters that delegate to the SearchUseCase.
  */
+/// Hard cap on the search page size. The default is 100; without a ceiling a
+/// client could pass `?limit=<huge>`, which flows straight into the SQL `LIMIT`
+/// and would pull that many rows into memory (and into the result cache). 500
+/// is a generous page for a search UI — `total_count` still reflects the full
+/// match set, so deeper results stay reachable via `offset`. Mirrors the
+/// suggestions endpoint, which already clamps with `.min(20)`.
+const MAX_SEARCH_LIMIT: usize = 500;
+
 pub struct SearchHandler;
 
 impl SearchHandler {
@@ -62,7 +70,7 @@ impl SearchHandler {
             max_size: params.max_size,
             folder_id: params.folder_id,
             recursive: params.recursive.unwrap_or(true),
-            limit: params.limit.unwrap_or(100),
+            limit: params.limit.unwrap_or(100).min(MAX_SEARCH_LIMIT),
             offset: params.offset.unwrap_or(0),
             sort_by: params.sort_by.unwrap_or_else(|| "relevance".to_string()),
         };

src/interfaces/api/handlers/webdav_handler.rs

@@ -63,10 +63,23 @@ fn encode_path_segment(segment: &str) -> String {
 
 /// Percent-encode a full slash-separated path, encoding each segment individually.
 pub(crate) fn encode_uri_path(path: &str) -> String {
-    path.split('/')
+    use std::fmt::Write as _;
-        .map(encode_path_segment)
+    // `utf8_percent_encode` returns a `Display` adapter, so write each encoded
-        .collect::<Vec<_>>()
+    // segment straight into `out` — avoids a String per segment and the joined
-        .join("/")
+    // Vec the previous `.map(...).collect::<Vec<_>>().join("/")` allocated on
+    // every PROPFIND href.
+    let mut out = String::with_capacity(path.len() + 8);
+    for (i, segment) in path.split('/').enumerate() {
+        if i > 0 {
+            out.push('/');
+        }
+        let _ = write!(
+            out,
+            "{}",
+            utf8_percent_encode(segment, PATH_SEGMENT_ENCODE_SET)
+        );
+    }
+    out
 }
 
 /// Build the `<D:href>` value for a non-collection (file) resource.

src/interfaces/api/routes.rs

@@ -9,7 +9,7 @@ use axum::{
 };
 use serde_json::json;
 use std::sync::Arc;
-use tower_http::{compression::CompressionLayer, trace::TraceLayer};
+use tower_http::trace::TraceLayer;
 use utoipa::OpenApi;
 
 /// Liveness probe — returns 200 if the process is running, no DB check.
@@ -579,10 +579,10 @@ pub fn create_api_routes(app_state: &Arc<AppState>) -> Router<Arc<AppState>> {
         .with_state(app_state.clone());
     router = router.nest("/users", users_router);
 
-    // Transparent compression (gzip + brotli) for all API responses.
+    // Compression is applied once, globally, in `main.rs` with a content-type
-    // tower-http negotiates via Accept-Encoding and skips already-compressed
+    // aware predicate that skips already-compressed media. Re-applying it here
-    // content types automatically. No manual compression in handlers.
+    // would double-wrap `/api`: this inner layer (no predicate) would compress
-    router
+    // media downloads, burning CPU for ~0 gain and stripping `Content-Length`.
-        .layer(CompressionLayer::new().br(true).gzip(true))
+    // So this router only adds tracing; compression is the global layer's job.
-        .layer(TraceLayer::new_for_http())
+    router.layer(TraceLayer::new_for_http())
 }

src/main.rs

@@ -541,25 +541,70 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     app = app.layer(DefaultBodyLimit::max(BODY_LIMIT));
 
     // ── HTTP compression (gzip + Brotli) ─────────────────────────────────
-    // Negotiates the best encoding via Accept-Encoding.  Skips responses
+    // Negotiates the best encoding via Accept-Encoding. Policy: compress
-    // that are already compressed or wouldn't benefit (images, video, etc.).
+    // everything by default so no shrinkable response is ever missed (text,
-    // Compatible with a future reverse proxy — if the proxy sees
+    // JSON, JS/CSS, XML, SVG, fonts ttf/otf, WASM…), and skip ONLY content
-    // `Content-Encoding` it will pass the response through untouched.
+    // that is already compressed — where a second pass burns CPU and adds
+    // latency for ~0 bytes saved.
+    //
+    // We deliberately do NOT blanket-exclude `image/*`: `image/svg+xml` is
+    // plain text and compresses ~70%, so the genuinely-compressed raster
+    // formats are listed individually instead, leaving SVG compressible.
+    //
+    // This is the single, global compression layer (the `/api` router used to
+    // add its own predicate-less one, which silently compressed media). It is
+    // reverse-proxy friendly: a proxy that sees `Content-Encoding` passes
+    // the response through untouched.
     {
         use tower_http::compression::CompressionLayer;
         use tower_http::compression::predicate::{NotForContentType, Predicate, SizeAbove};
 
         let predicate = SizeAbove::new(256)
             .and(NotForContentType::GRPC)
-            .and(NotForContentType::IMAGES)
             .and(NotForContentType::SSE)
-            .and(NotForContentType::const_new("application/octet-stream"))
+            // ── already-compressed raster images (SVG intentionally absent) ──
+            .and(NotForContentType::const_new("image/jpeg"))
+            .and(NotForContentType::const_new("image/png"))
+            .and(NotForContentType::const_new("image/gif"))
+            .and(NotForContentType::const_new("image/webp"))
+            .and(NotForContentType::const_new("image/avif"))
+            .and(NotForContentType::const_new("image/heic"))
+            .and(NotForContentType::const_new("image/heif"))
+            .and(NotForContentType::const_new("image/jp2"))
+            .and(NotForContentType::const_new("image/x-icon"))
+            .and(NotForContentType::const_new("image/vnd.microsoft.icon"))
+            // ── audio / video families (already compressed) ──
+            .and(NotForContentType::const_new("video/"))
+            .and(NotForContentType::const_new("audio/"))
+            // ── already-compressed web fonts; ttf/otf left compressible ──
+            .and(NotForContentType::const_new("font/woff"))
+            .and(NotForContentType::const_new("application/font-woff"))
+            // ── archives & compressed containers ──
             .and(NotForContentType::const_new("application/zip"))
             .and(NotForContentType::const_new("application/gzip"))
+            .and(NotForContentType::const_new("application/x-gzip"))
             .and(NotForContentType::const_new("application/x-tar"))
+            .and(NotForContentType::const_new("application/x-7z-compressed"))
+            .and(NotForContentType::const_new("application/x-rar-compressed"))
+            .and(NotForContentType::const_new("application/x-bzip2"))
+            .and(NotForContentType::const_new("application/zstd"))
+            .and(NotForContentType::const_new("application/x-xz"))
+            // ── zip-based document / app bundles (docx/xlsx/pptx, odf, epub…) ──
+            .and(NotForContentType::const_new(
+                "application/vnd.openxmlformats-officedocument",
+            ))
+            .and(NotForContentType::const_new(
+                "application/vnd.oasis.opendocument",
+            ))
+            .and(NotForContentType::const_new("application/epub+zip"))
+            .and(NotForContentType::const_new("application/java-archive"))
+            .and(NotForContentType::const_new(
+                "application/vnd.android.package-archive",
+            ))
+            // ── PDF: streams are usually already deflated; often large ──
             .and(NotForContentType::const_new("application/pdf"))
-            .and(NotForContentType::const_new("video/"))
+            // ── opaque binary we couldn't identify ──
-            .and(NotForContentType::const_new("audio/"));
+            .and(NotForContentType::const_new("application/octet-stream"));
 
         app = app.layer(CompressionLayer::new().compress_when(predicate));
     }