diff --git a/examples/bench_search_cache_mem.rs b/examples/bench_search_cache_mem.rs index 33d846da..e914e239 100644 --- a/examples/bench_search_cache_mem.rs +++ b/examples/bench_search_cache_mem.rs @@ -208,12 +208,20 @@ struct PhaseReport { /// Insert the full corpus, settle the cache, then measure retention and /// hot-key read latency. Identical for both variants — only the cache /// configuration differs. -async fn run_phase(cache: &moka::future::Cache>) -> PhaseReport { +async fn run_phase( + cache: &moka::future::Cache<(uuid::Uuid, u64), Arc>, +) -> PhaseReport { let hwm_start_kb = status_kb("VmHWM"); let rss_start_kb = status_kb("VmRSS"); + // Cache key changed to `(Uuid, u64)` in the per-user invalidation + // refactor (2026-07-26). Bench uses one fixed user across all keys — + // varying the u64 part exercises the same cardinality the pre-refactor + // benchmark did (one entry per query variant). + let bench_user = uuid::Uuid::nil(); + for i in 0..ENTRIES { - cache.insert(i, synth_entry(i)).await; + cache.insert((bench_user, i), synth_entry(i)).await; // Let eviction run as it would under live traffic, so evicted pages // are actually freed instead of piling up in moka's pending queue. if i % 64 == 0 { @@ -231,7 +239,7 @@ async fn run_phase(cache: &moka::future::Cache>) -> P .sum(); // Hot-key read latency: p50 over GETS reads of one resident key. - let hot: u64 = *cache.iter().next().expect("cache is empty after fill").0; + let hot: (uuid::Uuid, u64) = *cache.iter().next().expect("cache is empty after fill").0; for _ in 0..1_000 { black_box(cache.get(&hot).await); // warmup } @@ -262,7 +270,10 @@ async fn run_phase(cache: &moka::future::Cache>) -> P #[tokio::main] async fn main() { - let entry_weight = u64::from(search_results_entry_weight(&0, &synth_entry(0))); + let entry_weight = u64::from(search_results_entry_weight( + &(uuid::Uuid::nil(), 0), + &synth_entry(0), + )); println!("\n###########################################################"); println!("# Search-results cache: entry-count bound vs byte bound"); println!( @@ -279,7 +290,10 @@ async fn main() { println!("###########################################################\n"); // --- Phase 1: BEFORE (entry-count bound, exactly the old wiring) --- - let before_cache: moka::future::Cache> = + // Key type mirrors production's post-2026-07-26 tuple key so both + // phases exercise the same `Cache<(Uuid, u64), _>` shape; only the + // capacity bound differs (entry-count here vs weigher below). + let before_cache: moka::future::Cache<(uuid::Uuid, u64), Arc> = moka::future::Cache::builder() .max_capacity(BEFORE_MAX_ENTRIES) .time_to_live(Duration::from_secs(TTL_SECS)) diff --git a/src/application/services/favorites_service.rs b/src/application/services/favorites_service.rs index 5bb7a9d7..f166f998 100644 --- a/src/application/services/favorites_service.rs +++ b/src/application/services/favorites_service.rs @@ -11,6 +11,7 @@ use crate::application::dtos::favorites_dto::{ }; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::application::ports::favorites_ports::{FavoritesRepositoryPort, FavoritesUseCase}; +use crate::application::services::search_service::SearchService; use crate::common::errors::Result; use crate::domain::services::authorization::{Permission, Resource, ResourceKind, Subject}; use crate::infrastructure::repositories::pg::FavoritesPgRepository; @@ -29,14 +30,28 @@ pub struct FavoritesService { /// return name/mime/size/drive_id for any UUID the caller was /// able to enroll. See `docs/plan/authz_audit/rest_storage.md`. authorization: Arc, + /// Optional search-cache invalidator. Every favorite mutation + /// changes what `is_favorite` returns on the caller's cached + /// search result pages; without this hook the user sees a stale + /// star badge for up to the search cache's 5-minute TTL (Ed's + /// 2026-07-26 UX report). `None` when search is disabled + /// (`OXICLOUD_ENABLE_SEARCH=false`). + search: Option>, } impl FavoritesService { - /// Create a new FavoritesService with the given repository port - pub fn new(repo: Arc, authorization: Arc) -> Self { + /// Create a new FavoritesService with the given repository port. + /// `search` is `None` when search is disabled — the favorites path + /// still works, just without the cache-invalidation callback. + pub fn new( + repo: Arc, + authorization: Arc, + search: Option>, + ) -> Self { Self { repo, authorization, + search, } } @@ -103,6 +118,12 @@ impl FavoritesUseCase for FavoritesService { .await?; self.repo.add_favorite(user_id, item_id, item_type).await?; + // Drop this user's cached search pages so a subsequent search + // reflects the new star. Scoped to the caller — other tenants' + // caches are untouched. + if let Some(search) = &self.search { + search.invalidate_for_user(user_id).await; + } info!( "Successfully added {} '{}' to favorites for user {}", item_type, item_id, user_id @@ -125,6 +146,12 @@ impl FavoritesUseCase for FavoritesService { .repo .remove_favorite(user_id, item_id, item_type) .await?; + // Only invalidate when a row was actually removed — a no-op + // remove (item wasn't favorited) doesn't need to cold-start the + // cache. Keeps the "toggle a non-favorite" no-op cheap. + if removed && let Some(search) = &self.search { + search.invalidate_for_user(user_id).await; + } info!( "{} {} '{}' from favorites for user {}", if removed { @@ -181,6 +208,14 @@ impl FavoritesUseCase for FavoritesService { let requested = items.len(); let inserted = self.repo.add_favorites_batch(user_id, items).await?; let already_existed = requested as u64 - inserted; + // Any actual insert flips is_favorite for at least one row — + // invalidate. Skip when the batch was fully idempotent (every + // item was already favorited); no user-visible change. + if inserted > 0 + && let Some(search) = &self.search + { + search.invalidate_for_user(user_id).await; + } info!( "Batch favorites for user {}: {} requested, {} inserted, {} already existed", diff --git a/src/application/services/search_service.rs b/src/application/services/search_service.rs index 0196aeae..1356d454 100644 --- a/src/application/services/search_service.rs +++ b/src/application/services/search_service.rs @@ -72,7 +72,14 @@ pub struct SearchService { /// Keys span user × query × offset × limit, and each page holds up to 500 /// enriched rows (~500–900 B of owned Strings each) — an entry-count bound /// let hundreds of MB of result pages accumulate invisibly. - search_cache: moka::future::Cache>, + /// + /// Key is `(user_id, criteria_hash)` (not a single fused `u64`) so + /// `invalidate_for_user` can predicate on `k.0` — a per-user flush + /// runs when the user favorites/shares a file so their next search + /// sees the fresh `is_favorite` / `is_shared` flags instead of a + /// cache entry that hardened at compute-time (was up to 5 min stale + /// before 2026-07-26 — Ed reported the mismatch). + search_cache: moka::future::Cache<(Uuid, u64), Arc>, } // ─── Search-results cache (byte-bounded) ───────────────────────────────── @@ -88,7 +95,7 @@ pub struct SearchService { /// /// `pub` so `examples/bench_search_cache_mem.rs` can recompute retained /// bytes with the exact production formula. -pub fn search_results_entry_weight(_key: &u64, value: &Arc) -> u32 { +pub fn search_results_entry_weight(_key: &(Uuid, u64), value: &Arc) -> u32 { /// Fixed per-row overhead: struct scalars + one 24-B header per `String` /// field (12 on a file row, 4 on a folder row) + `Vec` slot + allocator /// slop. Deliberately a round upper-ish estimate — under-weighing is the @@ -132,11 +139,15 @@ pub fn search_results_entry_weight(_key: &u64, value: &Arc) -> pub fn build_search_results_cache( cache_ttl_secs: u64, max_bytes: u64, -) -> moka::future::Cache> { +) -> moka::future::Cache<(Uuid, u64), Arc> { moka::future::Cache::builder() .max_capacity(max_bytes) .weigher(search_results_entry_weight) .time_to_live(Duration::from_secs(cache_ttl_secs)) + // Required for `invalidate_entries_if` to actually match anything + // — without this the closure silently no-ops (per the + // `bug_moka_invalidate_entries_if_needs_opt_in` memo). + .support_invalidation_closures() .build() } @@ -270,10 +281,13 @@ impl SearchService { } /// Creates a cache key from the search criteria using zero-allocation hashing. - fn create_cache_key(criteria: &SearchCriteriaDto, user_id: &str) -> u64 { + /// Hash just the criteria — the caller pairs the returned `u64` with + /// the `Uuid` user_id to form the composite cache key `(Uuid, u64)`. + /// Split from the fused hash so `invalidate_for_user` can predicate + /// on the user side of the tuple without decoding the criteria. + fn create_cache_key(criteria: &SearchCriteriaDto) -> u64 { let mut hasher = std::collections::hash_map::DefaultHasher::new(); criteria.hash(&mut hasher); - user_id.hash(&mut hasher); hasher.finish() } @@ -654,13 +668,11 @@ impl SearchUseCase for SearchService { criteria: SearchCriteriaDto, user_id: Uuid, ) -> Result> { - // Stack-encode the UUID (36 ASCII bytes) instead of `to_string()` — the - // hasher sees the identical byte sequence, so the u64 key is unchanged, - // but the per-request heap `String` is gone (the fn doc even claims - // "zero-allocation hashing"). See benches/ROUND19.md §M5. - let mut user_id_buf = [0u8; uuid::fmt::Hyphenated::LENGTH]; - let user_id_str = user_id.hyphenated().encode_lower(&mut user_id_buf); - let cache_key = Self::create_cache_key(&criteria, user_id_str); + // Composite key: `(user_id, criteria_hash)`. Pairs the identity of + // the caller with the hash of the request so `invalidate_for_user` + // can drop just this user's entries when their favorites / shares + // change (see the `search_cache` field doc for the "why"). + let cache_key = (user_id, Self::create_cache_key(&criteria)); // Single-flight: collapse N identical concurrent searches into ONE // execution. `try_get_with` serves the cached result on a hit and, on a @@ -997,6 +1009,41 @@ impl SearchUseCase for SearchService { } } +impl SearchService { + /// Drop every cached search page for a single user. Called by the + /// favorites / share services after a mutation that changes what + /// `is_favorite` / `is_shared` would return for one of the caller's + /// files — without this the caller would see a stale flag for up + /// to `cache_ttl_secs` (Ed's 2026-07-26 report). + /// + /// `invalidate_entries_if` needs `.support_invalidation_closures()` + /// on the cache builder — set in `build_search_results_cache`. This + /// is scoped (predicate matches `k.0 == user_id` on the composite + /// `(Uuid, u64)` key), so a per-user favorite toggle does NOT + /// cold-start every other tenant's cache the way `invalidate_all` + /// does on the admin cache-flush endpoint. + pub async fn invalidate_for_user(&self, user_id: Uuid) { + // moka registers the predicate and returns a `PredicateId` — we + // don't need the id (we're not planning to unregister). Errors + // here are non-critical: worst case the caller sees stale + // is_favorite / is_shared for TTL seconds, exactly the state + // before this fix. Log-and-swallow keeps the mutation path + // reliable even under moka pressure. + if let Err(e) = self + .search_cache + .invalidate_entries_if(move |k, _| k.0 == user_id) + { + tracing::warn!( + target: "oxicloud::search", + error = %e, + %user_id, + "search cache invalidate_entries_if failed — user will see \ + stale is_favorite / is_shared until TTL expires", + ); + } + } +} + // ─── Stub for testing ──────────────────────────────────────────────────── impl SearchService { @@ -1080,7 +1127,7 @@ mod tests { fn entry_weight_counts_every_owned_string_plus_overheads() { // Empty page: entry overhead + sort_by ("relevance" = 9 bytes). let empty = Arc::new(SearchResultsDto::empty()); - let base = search_results_entry_weight(&0, &empty) as usize; + let base = search_results_entry_weight(&(Uuid::nil(), 0), &empty) as usize; assert_eq!(base, 256 + 9); // One file row: base + row overhead + its owned string bytes @@ -1094,7 +1141,7 @@ mod tests { 0, "relevance".to_string(), )); - let w = search_results_entry_weight(&0, &one_file) as usize; + let w = search_results_entry_weight(&(Uuid::nil(), 0), &one_file) as usize; assert_eq!(w, base + 200 + 7 + 7 + 8 + 10); // Folder rows weigh too (id 2 + name 4 + path 5 + parent 6 = 17). @@ -1122,7 +1169,7 @@ mod tests { 0, "relevance".to_string(), )); - let w = search_results_entry_weight(&0, &one_folder) as usize; + let w = search_results_entry_weight(&(Uuid::nil(), 0), &one_folder) as usize; assert_eq!(w, base + 200 + 2 + 4 + 5 + 6); } @@ -1143,12 +1190,12 @@ mod tests { "relevance".to_string(), )) }; - let per_entry = search_results_entry_weight(&0, &entry(0)) as u64; + let per_entry = search_results_entry_weight(&(Uuid::nil(), 0), &entry(0)) as u64; let budget = per_entry * 2 + per_entry / 2; let cache = build_search_results_cache(300, budget); for i in 0..20u64 { - cache.insert(i, entry(i as usize)).await; + cache.insert((Uuid::nil(), i), entry(i as usize)).await; } cache.run_pending_tasks().await; diff --git a/src/application/services/share_service.rs b/src/application/services/share_service.rs index 1beebf4c..9e9ac159 100644 --- a/src/application/services/share_service.rs +++ b/src/application/services/share_service.rs @@ -4,6 +4,7 @@ use thiserror::Error; use tokio::sync::Semaphore; use uuid::Uuid; +use crate::application::services::search_service::SearchService; use crate::domain::repositories::drive_repository::DriveRepository; use crate::domain::repositories::folder_repository::FolderRepository; use crate::domain::services::authorization::{Permission, Resource, Role, Subject}; @@ -98,6 +99,16 @@ pub struct ShareService { /// Bounds the number of in-flight Argon2 password hashes to avoid /// saturating the blocking thread pool and consuming excessive RAM. hash_semaphore: Arc, + /// Optional search-cache invalidator. Every share create/delete flips + /// what `is_shared` returns on the calling user's cached search + /// result pages; without this hook the sharer sees a stale share + /// badge for up to the search cache's 5-minute TTL. `None` when + /// search is disabled (`OXICLOUD_ENABLE_SEARCH=false`). + /// + /// Only the CALLER's cache is invalidated — recipients of a share + /// still get stale-until-TTL for now (would need a per-resource + /// invalidation index; deferred). + search: Option>, } impl ShareService { @@ -110,6 +121,7 @@ impl ShareService { drive_repository: Arc, password_hasher: Arc, authorization: Arc, + search: Option>, ) -> Self { Self { base_url: config.base_url(), @@ -121,6 +133,7 @@ impl ShareService { password_hasher, authorization, hash_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_HASHES)), + search, } } @@ -342,6 +355,13 @@ impl ShareUseCase for ShareService { .await .map_err(|e| ShareServiceError::Repository(e.to_string()))?; + // Sharer's search cache no longer reflects `is_shared` truthfully + // for the affected resource — flush their entries. Recipients are + // still stale-until-TTL (see the struct field comment). + if let Some(search) = &self.search { + search.invalidate_for_user(user_id).await; + } + // Return DTO with the requested expires_at (grant subquery on the share // row would return NULL at this point since INSERT ran before the grant). let mut response = ShareDto::from_entity(&saved_share, &self.base_url); @@ -445,6 +465,12 @@ impl ShareUseCase for ShareService { self.share_repository .delete_share_for_user(id, requester_id) .await?; + // Sharer's search cache no longer reflects `is_shared` truthfully + // for the affected resource — flush their entries. Recipients are + // still stale-until-TTL (see the struct field comment). + if let Some(search) = &self.search { + search.invalidate_for_user(requester_id).await; + } Ok(()) } diff --git a/src/common/di.rs b/src/common/di.rs index 440b39ce..e6d6dc27 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -877,13 +877,17 @@ impl AppServiceFactory { Some(service as Arc) } - /// Creates the sharing service + /// Creates the sharing service. `search_service` is threaded through + /// so create/delete of a share can flush the caller's cached search + /// pages (2026-07-26 — per-user is_shared invalidation). `None` when + /// search is disabled; the flush becomes a no-op. pub fn create_share_service( &self, repos: &RepositoryServices, db_pool: &Arc, authorization: &Arc, drive_repo: &Arc, + search_service: Option>, ) -> Option> { if !self.config.features.enable_file_sharing { tracing::info!("File sharing service is disabled in configuration"); @@ -909,6 +913,10 @@ impl AppServiceFactory { drive_repo.clone(), password_hasher, authorization.clone(), + // Optional per-user search-cache invalidator — set here so + // create/delete of a share drops the sharer's cached search + // pages (2026-07-26). `None` when search is disabled. + search_service.clone(), )); tracing::info!("File sharing service initialized"); @@ -917,16 +925,23 @@ impl AppServiceFactory { /// Creates the favorites service (requires database + authz engine /// for the Read gate on `add_to_favorites` — see the post-Drive - /// AuthZ audit). + /// AuthZ audit). `search_service` is threaded through so add/remove + /// can flush the caller's cached search pages (2026-07-26 — per-user + /// is_favorite invalidation). `None` when search is disabled. pub fn create_favorites_service( &self, db_pool: &Arc, authorization: &Arc, + search_service: Option>, ) -> Arc { let repo = Arc::new( crate::infrastructure::repositories::pg::FavoritesPgRepository::new(db_pool.clone()), ); - let service = Arc::new(FavoritesService::new(repo, authorization.clone())); + let service = Arc::new(FavoritesService::new( + repo, + authorization.clone(), + search_service, + )); tracing::info!("Favorites service initialized"); service } @@ -1325,7 +1340,13 @@ impl AppServiceFactory { ); // 5. Share service - let share_service = self.create_share_service(&repos, &pool, &authorization, &drive_repo); + let share_service = self.create_share_service( + &repos, + &pool, + &authorization, + &drive_repo, + apps.search_service.clone(), + ); apps.share_service = share_service.clone(); let share_browse_service = share_service.as_ref().map(|s| { @@ -1359,7 +1380,8 @@ impl AppServiceFactory { > = None; { - let favs = self.create_favorites_service(&pool, &authorization); + let favs = + self.create_favorites_service(&pool, &authorization, apps.search_service.clone()); favorites_service = Some(favs.clone()); apps.favorites_service = Some(favs);