diff --git a/docs/architecture/02-svelte-frontend.md b/docs/architecture/02-svelte-frontend.md index 93d2a8e06..dd4e9e83b 100644 --- a/docs/architecture/02-svelte-frontend.md +++ b/docs/architecture/02-svelte-frontend.md @@ -314,57 +314,22 @@ pub struct PlaybackModeManager { **Location**: `src-tauri/src/storage/db_service.rs` -Async database interface wrapping synchronous `rusqlite` to prevent blocking the Tokio runtime: +Async database interface over `rusqlite`. `RusqliteService` owns the +database: writes run as jobs on one writer thread, reads on a pool of read-only +WAL connections, so a read never waits for a write. Callers only see the +trait (`execute`, `insert`, `execute_detached`, `query_one` / `query_optional` / +`query_many`, `transaction`, `transaction_without_foreign_keys`) and build +queries with `Query` + `QueryParam`, which keeps values out of the SQL string. ```rust -#[async_trait] -pub trait DatabaseService: Send + Sync { - async fn execute(&self, query: Query) -> Result; - async fn execute_batch(&self, queries: Vec) -> Result<(), DatabaseError>; - async fn query_one(&self, query: Query, mapper: F) -> Result - where F: FnOnce(&Row) -> Result + Send + 'static; - async fn query_optional(&self, query: Query, mapper: F) -> Result, DatabaseError> - where F: FnOnce(&Row) -> Result + Send + 'static; - async fn query_many(&self, query: Query, mapper: F) -> Result, DatabaseError> - where F: Fn(&Row) -> Result + Send + 'static; - async fn transaction(&self, f: F) -> Result - where F: FnOnce(Transaction) -> Result + Send + 'static; -} - -pub struct RusqliteService { - connection: Arc>, -} - -impl DatabaseService for RusqliteService { - async fn execute(&self, query: Query) -> Result { - let conn = self.connection.clone(); - tokio::task::spawn_blocking(move || { - // Execute query on blocking thread pool - }).await? - } - // ... other methods use spawn_blocking -} -``` - -**Key Benefits:** -- **No Freezing**: All blocking DB ops run in thread pool via `spawn_blocking` -- **Type Safety**: `QueryParam` enum prevents SQL injection -- **Future Proof**: Easy to swap to native async DB (tokio-rusqlite) -- **Testable**: Can mock DatabaseService for tests - -**Usage Pattern:** -```rust -// Before (blocking - causes UI freeze) -let conn = database.connection(); -let conn = conn.lock().unwrap(); // BLOCKS -conn.query_row(...) // BLOCKS - -// After (async - no freezing) -let db_service = database.service(); +let db_service = database.service(); // cheap clone of the shared owner let query = Query::with_params("SELECT ...", vec![...]); -db_service.query_one(query, |row| {...}).await // spawn_blocking internally +db_service.query_one(query, |row| {...}).await // runs on a reader ``` +Ownership model, invariants and the reasons for them: +[08-database-design.md → Connection ownership](08-database-design.md#connection-ownership). + ## Component Hierarchy ```mermaid diff --git a/docs/architecture/03-data-flow.md b/docs/architecture/03-data-flow.md index 5bf790d64..cd0782575 100644 --- a/docs/architecture/03-data-flow.md +++ b/docs/architecture/03-data-flow.md @@ -28,12 +28,17 @@ sequenceDiagram Server->>Conn: mark_unreachable() (debounced) end - alt Cache returns with content + alt Cache answers first with content (inside 100ms, or later but before the server) Cache-->>Hybrid: Result with items Hybrid-->>Rust: Return cache result - else Cache timeout or empty + Server-->>Hybrid: Fresh result (later) + Hybrid->>Cache: save_to_cache() in background + else Server answers first, or cache is empty Server-->>Hybrid: Fresh result Hybrid-->>Rust: Return server result + else Server fails + Cache-->>Hybrid: Whatever the cache has (waited for) + Hybrid-->>Rust: Return cache result, else the server error end Rust-->>Client: SearchResult @@ -42,11 +47,21 @@ sequenceDiagram ``` **Key Points:** -- Cache queries have 100ms timeout for responsiveness -- Server queries always run for fresh data -- Cache wins if it has meaningful content -- Automatic fallback to server if cache is empty/stale -- Background cache updates (planned) +- Both legs start together. A cache answer with content inside 100 ms + (`CACHE_FAST_PATH`) returns at once. +- **The deadline does not decide the race.** A cache read still running at + 100 ms is raced against the server (`HybridRepository::race_slow_cache`), and + whichever answers first *with content* wins. It used to be that a read past + the deadline was only consulted if the server failed, so a page whose cache + read took 150 ms always paid the full server round trip — about a second on a + phone, on every visit. +- An empty or failed cache answer is not a win; the server decides. A failed + server falls back to whatever the cache said, waiting for it if necessary. +- On a cache win the server's page is still cached in the background when it + arrives, so per-user state (positions, favourites) keeps up. +- A `get_items` leg that takes 250 ms or more is logged at INFO with its row + count, so a slow page can be attributed to the cache or the server from a + device log alone. - **Connectivity side-effect**: each server request feeds the `ConnectivityMonitor`, which is the source of truth for the offline/online banner (see [07-connectivity.md](07-connectivity.md)). A server-answered error (401/404/5xx) still counts as *reachable* — only network failures, sustained past a debounce window, flip the app to offline. ### Listing order is decided in Rust diff --git a/docs/architecture/06-downloads-and-offline.md b/docs/architecture/06-downloads-and-offline.md index d3f8b2bbb..407d3227a 100644 --- a/docs/architecture/06-downloads-and-offline.md +++ b/docs/architecture/06-downloads-and-offline.md @@ -224,10 +224,11 @@ them: the current user — local path, direct play, item id as media source (a download names no source, so the server served its default, which carries the item's id) — and `HybridRepository::get_playback_info` consults it **first**. -- **A slow cache read is waited for, never discarded.** The cache is one SQLite - connection behind one mutex, so any write in progress (the catalog sync that - starts at every launch, a download finishing) pushes a read past the 100 ms fast - path. `get_items`, the library list, genres and playlist items used to discard +- **A slow cache read is waited for, never discarded.** Reads no longer queue + behind writes (see [Connection ownership](08-database-design.md#connection-ownership)), + but a big query, a cold page cache or a busy reader pool can still push a read + past the 100 ms fast path. Such a read is raced against the server rather + than set aside (see [03-data-flow.md](03-data-flow.md)). `get_items`, the library list, genres and playlist items used to discard such a read, wait for the server, and — offline — return its error over data on disk; "More info" on a downloaded show failed that way. They now start the read with `cache_try` (which keeps it running) and `settle` on it when the server diff --git a/docs/architecture/08-database-design.md b/docs/architecture/08-database-design.md index cf793ea4a..9de4982ca 100644 --- a/docs/architecture/08-database-design.md +++ b/docs/architecture/08-database-design.md @@ -594,6 +594,99 @@ flowchart LR | Episode | ~3 KB | ~100 KB | 300 MB - 2 GB | | Full music library (5000 songs) | ~10 MB | ~250 MB | 25-75 GB | +## Connection ownership + +`storage::Database` is opened once at startup and owns the database for the +life of the app; nothing else opens the file. Everything goes through +`Database::service()`, which returns a clone of one `RusqliteService` +(`storage/db_service.rs`) — callers never hold a `Connection`. + +``` + ┌─────────────────────────────┐ + execute / insert / │ writer thread ("db-writer") │ one read-write connection, + transaction / ────►│ jobs run in arrival order │ foreign_keys = ON + execute_detached └─────────────────────────────┘ + ┌─────────────────────────────┐ + query_one / │ reader pool (3 connections) │ query_only = ON; WAL gives + query_optional / ─►│ via spawn_blocking │ each the last committed + query_many └─────────────────────────────┘ snapshot +``` + +Pragmas: `journal_mode = WAL`, `synchronous = NORMAL`, `busy_timeout = 5 s` on +every connection. + +**Why.** It used to be one connection behind one `std::sync::Mutex`. WAL was on, +but with a single connection its one benefit — readers running beside a writer — +never applied: every read queued behind every write. A series page's background +refresh cached hundreds of episodes one autocommit (and one fsync) at a time, +and library pages, thumbnail lookups and settings reads all waited behind the +pile; the same page measured 1.7 s or 3.95 s depending on what was queued. + +**Invariants a change must keep:** + +- **Writes go to the writer, reads to the pool.** A reader is `query_only`, so a + write sent through `query_*` fails loudly rather than racing the writer. A + read that must see a write *in the same unit of work* belongs inside the + `transaction` closure, which runs on the writer. +- **Connection-wide state is set inside one writer job.** `PRAGMA foreign_keys` + is per connection and ignored inside a transaction. `save_to_cache` used to + switch it off with one `execute` and back on with another, so it stayed off + for every other write that ran across the save's awaits (they did — nearly + all of 1,600 FK-violating writes got through in the regression test). + `transaction_without_foreign_keys` flips it around `BEGIN`/`COMMIT` in a single + job. Any future pragma toggle must work the same way. +- **An insert's rowid comes from the same job.** `insert()` returns it; there is + deliberately no standalone `last_insert_rowid()`, which returned whichever row + the last *anyone* inserted. +- **Batch writes into one transaction.** A commit is a queue slot on the + writer; a page of items is one `transaction`, not one `execute` per row. +- **Best-effort bookkeeping does not wait.** `execute_detached` queues a write + (the thumbnail LRU access time) in order with the rest and returns + immediately; failures are only logged. +- **The writer survives a panicking job.** The job's caller gets an error; the + loop rolls back any transaction the job left open and restores + `foreign_keys = ON`, then carries on. + +**`synchronous = NORMAL`** is corruption-safe in WAL mode and survives an app +crash; only a power cut can lose the last few commits. Everything stored is +either re-fetchable from the server or (the sync queue, local positions) +recoverable at that granularity. + +**In-memory databases** (tests) cannot be shared between connections, so +`RusqliteService::new` has no pool and routes reads through the writer — the +old serialized behaviour. + +**Not done, and why:** grouping consecutive small `execute` jobs into one +commit on the writer. With `synchronous = NORMAL` a commit no longer fsyncs, so +the gain is small, and it would change a failed statement's error semantics +for the jobs grouped with it. + +### Listing query shape + +`OfflineRepository::get_items` (`items_listing_sql`) is the hot read: every +library, series, season and album page goes through it, and it must fit the +100 ms cache fast path on a phone. + +- **Availability is checked per row, with `EXISTS`** (cached for browsing, + downloaded, or a container with a downloaded child). It used to be a CTE + that built the id of *every* available item in the database before filtering + to the parent: ~80 ms on a desktop for a 100k-item cache, whatever the parent. +- **A parent that is not a library is matched on the hierarchy columns alone** + (`parent_id`/`album_id`/`season_id`/`series_id`), which SQLite answers with a + multi-index `OR`. Whether the parent is a library is looked up first; the + library clause can only match for a library, and inside the same `OR` it + forced a full scan. +- **`+i.server_id`**: the app never runs `ANALYZE`, and without statistics the + planner prefers the `server_id` index — which every row shares — over the + hierarchy indexes. The unary `+` takes it out of consideration. +- **User data is fetched in batches** (`with_user_data`, one `IN (…)` query per + 500 rows), not once per row. + +Result: 80 ms → 1.5 ms on the desktop benchmark; +`listing_a_non_library_parent_uses_the_hierarchy_indexes` asserts the plan. +`storage::tests::write_bench_database` (ignored) writes a phone-sized +catalogue for timing queries with the `sqlite3` CLI. + ## Rust Module Structure ``` diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index e64f68cf5..328313ec9 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -2315,7 +2315,6 @@ dependencies = [ "tempfile", "tiny_http", "tokio", - "tokio-rusqlite", "tokio-util", "urlencoding", "uuid", @@ -5340,17 +5339,6 @@ dependencies = [ "syn 2.0.112", ] -[[package]] -name = "tokio-rusqlite" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b65501378eb676f400c57991f42cbd0986827ab5c5200c53f206d710fb32a945" -dependencies = [ - "crossbeam-channel", - "rusqlite", - "tokio", -] - [[package]] name = "tokio-rustls" version = "0.26.4" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index fb8b50edc..ae4142d9a 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -53,7 +53,6 @@ futures-util = "0.3" async-trait = "0.1" # SQLite for offline storage -tokio-rusqlite = "0.6" rusqlite = { version = "0.32", features = ["bundled"] } chrono = { version = "0.4", features = ["serde"] } directories = "5" diff --git a/src-tauri/src/commands/sync.rs b/src-tauri/src/commands/sync.rs index c8233c506..0e4b3b110 100644 --- a/src-tauri/src/commands/sync.rs +++ b/src-tauri/src/commands/sync.rs @@ -49,6 +49,19 @@ pub async fn sync_queue_mutation( Arc::new(database.service()) }; + enqueue_mutation(&*db_service, user_id, operation, item_id, payload).await +} + +/// Insert one pending mutation and return the id of *that* row. +/// +/// TRACES: UR-002, UR-017 | DR-014 +pub(crate) async fn enqueue_mutation( + db_service: &S, + user_id: String, + operation: String, + item_id: Option, + payload: Option, +) -> Result { let query = Query::with_params( "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at) VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP)", @@ -60,13 +73,9 @@ pub async fn sync_queue_mutation( ], ); - db_service.execute(query).await.map_err(|e| e.to_string())?; - let id = db_service - .last_insert_rowid() - .await - .map_err(|e| e.to_string())?; - - Ok(id) + // `insert`, not `execute` + `last_insert_rowid`: the id must be read in + // the same job as the insert, or a concurrent write hands us its row. + db_service.insert(query).await } /// Get all pending sync operations for a user @@ -387,4 +396,58 @@ mod tests { assert!(item.retry_count == i); } } + + /// Each queued mutation must get back the id of its *own* row. + /// + /// The id used to come from a separate `last_insert_rowid()` call — a + /// second trip to the shared connection — so another insert landing in + /// between handed this mutation someone else's id, and marking it synced + /// later completed the wrong row. + /// + /// TRACES: UR-002, UR-017 | DR-014 | UT-014 + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] + async fn concurrent_enqueues_each_get_their_own_row_id() { + let database = crate::storage::Database::open_in_memory().unwrap(); + let service = Arc::new(database.service()); + service + .execute(Query::new( + "INSERT INTO servers (id, name, url) VALUES ('s', 'S', 'http://s')", + )) + .await + .unwrap(); + service + .execute(Query::new( + "INSERT INTO users (id, server_id, username) VALUES ('u', 's', 'u')", + )) + .await + .unwrap(); + + let tasks: Vec<_> = (0..200) + .map(|i| { + let service = Arc::clone(&service); + tokio::spawn(async move { + let op = format!("op-{i}"); + let id = enqueue_mutation(&*service, "u".into(), op.clone(), None, None) + .await + .unwrap(); + (op, id) + }) + }) + .collect(); + + for task in tasks { + let (op, id) = task.await.unwrap(); + let stored: String = service + .query_one( + Query::with_params( + "SELECT operation FROM sync_queue WHERE id = ?", + vec![QueryParam::Int64(id)], + ), + |row| row.get(0), + ) + .await + .unwrap(); + assert_eq!(stored, op, "mutation {op} was handed row {id}"); + } + } } diff --git a/src-tauri/src/repository/hybrid.rs b/src-tauri/src/repository/hybrid.rs index dc671a1ee..a4c71b04f 100644 --- a/src-tauri/src/repository/hybrid.rs +++ b/src-tauri/src/repository/hybrid.rs @@ -50,6 +50,42 @@ impl CacheLeg { } } +/// Log a `get_items` leg that took long enough to be felt, so a slow page can +/// be attributed to the cache or the server from a device log alone. +fn log_slow_leg( + leg: &str, + parent_id: &str, + started: std::time::Instant, + result: &Result, +) { + let ms = started.elapsed().as_millis(); + let outcome = match result { + Ok(data) => format!("{} rows", data.items.len()), + Err(e) => format!("error: {e:?}"), + }; + let parent = &parent_id[..8.min(parent_id.len())]; + if ms >= 250 { + log::info!("[HybridRepo] get_items {leg} leg for {parent} took {ms} ms ({outcome})"); + } else { + debug!("[HybridRepo] get_items {leg} leg for {parent} took {ms} ms ({outcome})"); + } +} + +/// Outcome of racing a cache read that missed the fast path against the +/// server. See [`HybridRepository::race_slow_cache`]. +enum Raced { + /// The cache answered first with something to show (exclusions applied). + Cache(T), + /// The server answered first, or the cache had nothing. Raw: the caller + /// caches the full page and applies exclusions to what it returns. + Server(T), + /// The server failed; this is what the cache said instead (exclusions + /// applied), possibly an empty listing — which still beats an error. + Fallback(T), + /// Neither could answer; the server's error. + Failed(RepoError), +} + /// Hybrid repository combining online and offline data sources /// /// Uses cache-first parallel racing strategy: @@ -397,6 +433,64 @@ impl HybridRepository { } } + /// Race a cache read that missed the fast path against the server: + /// whichever answers first *with something to show* wins. + /// + /// The fast-path deadline bounds how long a cache hit may delay the UI; it + /// must not also decide the race. It used to: past 100 ms the query waited + /// for the server even when the cache answered moments later, so a page + /// whose cache read took 150 ms always paid the full server round trip + /// (about a second on a phone). An empty or failed cache answer is not a + /// win — the server decides then — and a failed server falls back to + /// whatever the cache said. + /// + /// TRACES: UR-002 | DR-013 + async fn race_slow_cache( + mut slow: tokio::task::JoinHandle>, + server: F, + ) -> Raced + where + T: MeaningfulContent + ExcludeHidden, + F: std::future::Future>, + { + tokio::pin!(server); + // `biased`, server first: if both are ready at once, the fresher + // answer wins. + tokio::select! { + biased; + server_result = &mut server => match server_result { + Ok(data) => Raced::Server(data), + Err(e) => { + debug!("[HybridRepo] Server failed; waiting for the slow cache query"); + match slow.await { + Ok(Ok(data)) => Raced::Fallback(data.without_excluded()), + _ => Raced::Failed(e), + } + } + }, + cache_result = &mut slow => { + let cache = cache_result + .unwrap_or_else(|join| Err(RepoError::Database { + message: format!("Cache query failed: {join}"), + })) + .map(ExcludeHidden::without_excluded); + match cache { + Ok(data) if data.has_content() => { + debug!("[HybridRepo] Slow cache answered before the server"); + Raced::Cache(data) + } + other => match server.await { + Ok(data) => Raced::Server(data), + Err(e) => match other { + Ok(data) => Raced::Fallback(data), + Err(_) => Raced::Failed(e), + }, + }, + } + } + } + } + /// Cache-first query: try cache, fall back to server on miss. /// /// 1. Check cache (100ms fast path, via `cache_leg`; a slow read keeps running) @@ -459,33 +553,28 @@ impl HybridRepository { } } - debug!("[HybridRepo] Cache miss or slow, querying server"); + // Still running: race it against the server rather than waiting the + // server out. If the server then fails, the cache is the only thing + // that can answer — offline, a cache read slowed by a concurrent write + // used to surface as a network error. + if let Some(handle) = slow { + return match Self::race_slow_cache(handle, server_future).await { + Raced::Cache(data) => { + on_cache_hit(); + Ok(data) + } + Raced::Server(data) => Ok(data.without_excluded()), + Raced::Fallback(data) => Ok(data), + Raced::Failed(e) => Err(e), + }; + } + + debug!("[HybridRepo] Cache miss, querying server"); match server_future.await { Ok(data) => Ok(data.without_excluded()), - Err(e) => { - // The server cannot answer. If the cache is still working, it is - // now the only thing that can, so wait it out rather than - // reporting the server's failure over data we are about to hold. - // This is the offline path: a cache read slowed by a concurrent - // write used to surface as a network error. - if let Some(handle) = slow { - debug!("[HybridRepo] Server failed; waiting for the slow cache query"); - return match handle.await { - Ok(Ok(data)) => Ok(data.without_excluded()), - Ok(Err(cache_err)) => { - debug!("[HybridRepo] Slow cache query also failed: {cache_err}"); - Err(e) - } - Err(join) => { - debug!("[HybridRepo] Slow cache query panicked: {join}"); - Err(e) - } - }; - } - // Cache answered in time but had nothing: return that, so an - // empty-but-valid cached listing still beats a network error. - fast.unwrap_or(Err(e)) - } + // Cache answered in time but had nothing: return that, so an + // empty-but-valid cached listing still beats a network error. + Err(e) => fast.unwrap_or(Err(e)), } } @@ -498,10 +587,10 @@ impl HybridRepository { /// /// Missing the deadline does **not** cancel the query — it keeps running on /// its own task and [`CacheLeg::settle`] can still collect it. That - /// distinction is the whole point. The database is one SQLite connection - /// behind one mutex, so a concurrent write (a sync drain, a bulk - /// `save_to_cache`) blocks reads for its duration and this deadline trips - /// routinely on slow storage. Treating that as "the cache is empty" while + /// distinction is the whole point. A read can miss the deadline for many + /// reasons — a large listing, slow storage, a busy reader pool (and, before + /// reads had their own connections, any write in progress). Treating that + /// as "the cache is empty" while /// throwing the answer away meant that offline — where the server leg also /// fails — browsing surfaced a network error instead of the cached content /// sitting right there on disk. @@ -532,9 +621,9 @@ impl HybridRepository { /// if it made it, and otherwise the read itself, still running. /// /// The cache-then-server queries used to *discard* a read that missed the - /// deadline. The database is one SQLite connection behind one mutex, so any - /// write in progress — the catalog sync that starts at every launch, a - /// download finishing — pushes a read past 100 ms routinely; offline the + /// deadline. When reads shared one connection with writes, any write in + /// progress — the catalog sync that starts at every launch, a download + /// finishing — pushed a read past 100 ms routinely; offline the /// server then failed too, and the page reported a network error over data /// sitting on disk. Keeping the read lets [`Self::settle`] wait for it. /// @@ -576,6 +665,48 @@ impl HybridRepository { fast.or(Err(server_err)) } + /// Cache the server's page once it arrives, without holding up the + /// caller, who has already answered from the cache. A failed or empty + /// server answer leaves the existing cache alone. + fn save_when_server_answers( + server: tokio::task::JoinHandle>, + offline: Arc, + parent_id: String, + ) { + tokio::spawn(async move { + if let Ok(Ok(server_data)) = server.await { + Self::save_in_background(offline, parent_id, &server_data); + } + }); + } + + /// Cache a server page in the background (one transaction; see + /// `OfflineRepository::save_to_cache`). + fn save_in_background( + offline: Arc, + parent_id: String, + server_data: &SearchResult, + ) { + if server_data.items.is_empty() { + return; + } + let items = server_data.items.clone(); + tokio::spawn(async move { + match offline.save_to_cache(&parent_id, &items).await { + Ok(_) => debug!( + "[HybridRepo] Cached {} items for parent {}", + items.len(), + &parent_id[..8.min(parent_id.len())] + ), + Err(e) => warn!( + "[HybridRepo] Failed to cache {} items: {:?}", + items.len(), + e + ), + } + }); + } + /// [`Self::settle`] for `get_items`, whose slow read has not yet had /// exclusions applied. async fn cache_or( @@ -653,8 +784,12 @@ impl MediaRepository for HybridRepository { let opts_clone = options.clone(); // Start server request in background (non-blocking) - let server_handle = - tokio::spawn(async move { online.get_items(&parent_id_clone, options).await }); + let mut server_handle = tokio::spawn(async move { + let started = std::time::Instant::now(); + let result = online.get_items(&parent_id_clone, options).await; + log_slow_leg("server", &parent_id_clone, started, &result); + result + }); // Check cache first (fast, 100ms timeout). // @@ -671,8 +806,13 @@ impl MediaRepository for HybridRepository { // is what made a downloaded show fail offline ("Failed to load item") // whenever a write held the database past 100 ms — the catalog sync // that starts at every launch does, routinely. TRACES: UR-002 | DR-294 - let (cache_result, slow_cache) = - Self::cache_try(async move { offline.get_items(&parent_id, opts_clone).await }).await; + let (cache_result, slow_cache) = Self::cache_try(async move { + let started = std::time::Instant::now(); + let result = offline.get_items(&parent_id, opts_clone).await; + log_slow_leg("cache", &parent_id, started, &result); + result + }) + .await; let cache_result = cache_result.map(ExcludeHidden::without_excluded); // Downloads-only gate: when the "Show all server media" toggle is off @@ -702,54 +842,45 @@ impl MediaRepository for HybridRepository { "[HybridRepo] Cache hit for get_items, returning immediately for parent {}", &parent_id_for_save[..8.min(parent_id_for_save.len())] ); - // Background: save server result to cache when it arrives - tokio::spawn(async move { - match server_handle.await { - Ok(Ok(server_data)) if !server_data.items.is_empty() => { - if let Err(e) = offline_for_save - .save_to_cache(&parent_id_for_save, &server_data.items) - .await - { - warn!("[HybridRepo] Background cache update failed: {:?}", e); - } else { - debug!( - "[HybridRepo] Background updated {} cached items for parent {}", - server_data.items.len(), - &parent_id_for_save[..8.min(parent_id_for_save.len())] - ); - } - } - _ => {} // Server failed or returned empty — keep existing cache - } - }); + Self::save_when_server_answers(server_handle, offline_for_save, parent_id_for_save); return Ok(data.clone()); } } - // Cache miss — wait for server result + // The cache missed the fast path but is still reading: race it against + // the server instead of waiting the server out (DR-013). + if let Some(slow) = slow_cache { + let server = async { + (&mut server_handle).await.unwrap_or_else(|join| { + Err(RepoError::Network { + message: format!("Server task failed: {}", join), + }) + }) + }; + return match Self::race_slow_cache(slow, server).await { + Raced::Cache(data) => { + // The server is still in flight: cache its answer when it + // lands, exactly as on a fast-path hit. + Self::save_when_server_answers( + server_handle, + offline_for_save, + parent_id_for_save, + ); + Ok(data) + } + Raced::Server(server_data) => { + Self::save_in_background(offline_for_save, parent_id_for_save, &server_data); + Ok(server_data.without_excluded()) + } + Raced::Fallback(data) => Ok(data), + Raced::Failed(e) => Err(e), + }; + } + + // Cache answered in time with nothing — wait for the server. match server_handle.await { Ok(Ok(server_data)) => { - if !server_data.items.is_empty() { - let items_clone = server_data.items.clone(); - tokio::spawn(async move { - if let Err(e) = offline_for_save - .save_to_cache(&parent_id_for_save, &items_clone) - .await - { - warn!( - "[HybridRepo] Failed to save {} items to cache: {:?}", - items_clone.len(), - e - ); - } else { - debug!( - "[HybridRepo] Saved {} items to cache for parent {}", - items_clone.len(), - &parent_id_for_save[..8.min(parent_id_for_save.len())] - ); - } - }); - } + Self::save_in_background(offline_for_save, parent_id_for_save, &server_data); // The cache keeps the server's full page (above) — an exclusion // is a view preference and can be undone, so hiding items from // the *cache* would make un-hiding them require a re-crawl. Only @@ -757,12 +888,12 @@ impl MediaRepository for HybridRepository { // TRACES: UR-076 | DR-209 Ok(server_data.without_excluded()) } - Ok(Err(e)) => Self::cache_or(cache_result, slow_cache, e).await, + Ok(Err(e)) => Self::cache_or(cache_result, None, e).await, Err(join_err) => { let e = RepoError::Network { message: format!("Server task failed: {}", join_err), }; - Self::cache_or(cache_result, slow_cache, e).await + Self::cache_or(cache_result, None, e).await } } } @@ -771,9 +902,9 @@ impl MediaRepository for HybridRepository { /// background so the stored copy keeps up with the server. /// /// The background refresh is what carries per-user state home: caching an - /// item runs `mirror_user_data`, which is the only path by which a watch - /// position set on another device reaches the local `user_data` row the - /// resume check reads. Without it a cache hit returned this device's own + /// item also writes its `user_data_mirror_query` row, the only path by + /// which a watch position set on another device reaches the local + /// `user_data` row the resume check reads. Without it a cache hit returned this device's own /// stale position forever and cross-device resume silently did nothing — /// `get_items` already refreshes this way, so browsing a season worked /// while opening the episode directly did not. @@ -1385,9 +1516,9 @@ mod tests { /// Offline, a cache read slowed past the fast path must still answer. /// - /// The database is one SQLite connection behind one mutex, so a concurrent - /// write blocks reads for its duration and the 100 ms fast path trips on - /// slow storage. The deadline used to *cancel* the read and report it as a + /// The 100 ms fast path trips routinely on slow storage (and, while reads + /// shared one connection with writes, behind any write). The deadline used + /// to *cancel* the read and report it as a /// miss; with the server leg also failing (offline), the user got a network /// error over cached content that was sitting on disk. /// @@ -1464,6 +1595,101 @@ mod tests { assert!(matches!(err, RepoError::Network { .. }), "got {err:?}"); } + /// A cache read that misses the fast path but lands before the server must + /// be what the user sees. + /// + /// The 100 ms deadline used to decide the race outright: past it, the page + /// waited for the server even when the cache answered a few milliseconds + /// later — on a phone, a series page showed its seasons after the ~1 s + /// server round trip instead of the ~150 ms cache read, on every visit. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn a_slow_cache_that_beats_the_server_wins() { + let cache = HybridRepository::cache_leg(async { + tokio::time::sleep(Duration::from_millis(150)).await; + Ok(vec![MediaItem { + id: "slow-cache".to_string(), + ..Default::default() + }]) + }) + .await; + + let server = async { + tokio::time::sleep(Duration::from_millis(1500)).await; + Ok(vec![MediaItem { + id: "server".to_string(), + ..Default::default() + }]) + }; + + let started = std::time::Instant::now(); + let got = HybridRepository::parallel_race(cache, server) + .await + .unwrap(); + assert_eq!( + got[0].id, "slow-cache", + "waited for the server over a cache answer" + ); + assert!(started.elapsed() < Duration::from_millis(1000)); + } + + /// A server that beats a slow cache still answers first. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn a_server_that_beats_a_slow_cache_wins() { + let cache = HybridRepository::cache_leg(async { + tokio::time::sleep(Duration::from_millis(1500)).await; + Ok(vec![MediaItem { + id: "slow-cache".to_string(), + ..Default::default() + }]) + }) + .await; + + let server = async { + tokio::time::sleep(Duration::from_millis(150)).await; + Ok(vec![MediaItem { + id: "server".to_string(), + ..Default::default() + }]) + }; + + let started = std::time::Instant::now(); + let got = HybridRepository::parallel_race(cache, server) + .await + .unwrap(); + assert_eq!(got[0].id, "server"); + assert!(started.elapsed() < Duration::from_millis(1000)); + } + + /// A slow cache that comes back *empty* is not an answer: the server + /// decides. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn an_empty_slow_cache_defers_to_the_server() { + let cache = HybridRepository::cache_leg(async { + tokio::time::sleep(Duration::from_millis(150)).await; + Ok(Vec::::new()) + }) + .await; + + let server = async { + tokio::time::sleep(Duration::from_millis(400)).await; + Ok(vec![MediaItem { + id: "server".to_string(), + ..Default::default() + }]) + }; + + let got = HybridRepository::parallel_race(cache, server) + .await + .unwrap(); + assert_eq!(got[0].id, "server"); + } + /// Mock offline repository that tracks queries and saves struct MockOfflineRepo { items: Arc>>, diff --git a/src-tauri/src/repository/offline.rs b/src-tauri/src/repository/offline.rs index d78e56fb9..aa0f771e9 100644 --- a/src-tauri/src/repository/offline.rs +++ b/src-tauri/src/repository/offline.rs @@ -235,23 +235,201 @@ impl OfflineRepository { ); self.db_service - .query_optional(query, |row| { - let playback_position_ticks: Option = row.get(0).ok(); - Ok(UserData { - playback_position_ticks, - playback_position_ms: playback_position_ticks.map(crate::domain::ticks_to_ms), - is_played: row.get::<_, Option>(1).ok().flatten().map(|v| v != 0), - is_favorite: row.get::<_, Option>(2).ok().flatten().map(|v| v != 0), - play_count: row.get(3).ok(), - last_played_date: row.get(4).ok(), - playback_context_type: row.get(5).ok(), - playback_context_id: row.get(6).ok(), - }) - }) + .query_optional(query, |row| Ok(row_to_user_data(row, 0))) .await .ok() .flatten() } + + /// Attach each cached row's user data, fetched in batches. + /// + /// Listings used to call `get_user_data` once per row — a series page's + /// ~275 cached rows meant ~275 sequential database round trips, which alone + /// pushed the read past the cache fast path so the page waited for the + /// server every time. One `IN (…)` query per chunk instead; the chunk stays + /// well under SQLite's bound-parameter limit. + /// + /// TRACES: UR-002 | DR-013 + async fn with_user_data(&self, cached_items: Vec) -> Vec { + const CHUNK: usize = 500; + let mut by_id: std::collections::HashMap = + std::collections::HashMap::new(); + + for chunk in cached_items.chunks(CHUNK) { + let placeholders = vec!["?"; chunk.len()].join(","); + let mut params = vec![QueryParam::String(self.user_id.clone())]; + params.extend(chunk.iter().map(|c| QueryParam::String(c.id.clone()))); + let query = Query::with_params( + format!( + "SELECT item_id, playback_position_ticks, is_played, is_favorite, play_count, + last_played_at, playback_context_type, playback_context_id + FROM user_data WHERE user_id = ? AND item_id IN ({placeholders})" + ), + params, + ); + match self + .db_service + .query_many(query, |row| { + Ok((row.get::<_, String>(0)?, row_to_user_data(row, 1))) + }) + .await + { + Ok(rows) => by_id.extend(rows), + // Same as the per-row lookup: missing user data is not fatal. + Err(e) => debug!("[OfflineRepo] user_data batch lookup failed: {}", e), + } + } + + cached_items + .into_iter() + .map(|cached| { + let user_data = by_id.remove(&cached.id); + Self::cached_item_to_media_item(cached, user_data) + }) + .collect() + } +} + +/// A `user_data` row's columns, starting at `offset`, in the order +/// `playback_position_ticks, is_played, is_favorite, play_count, +/// last_played_at, playback_context_type, playback_context_id`. +fn row_to_user_data(row: &rusqlite::Row, offset: usize) -> UserData { + let playback_position_ticks: Option = row.get(offset).ok(); + UserData { + playback_position_ticks, + playback_position_ms: playback_position_ticks.map(crate::domain::ticks_to_ms), + is_played: row + .get::<_, Option>(offset + 1) + .ok() + .flatten() + .map(|v| v != 0), + is_favorite: row + .get::<_, Option>(offset + 2) + .ok() + .flatten() + .map(|v| v != 0), + play_count: row.get(offset + 3).ok(), + last_played_date: row.get(offset + 4).ok(), + playback_context_type: row.get(offset + 5).ok(), + playback_context_id: row.get(offset + 6).ok(), + } +} + +/// The listing query behind `get_items`: the cached children of one parent +/// that are available to show. +/// +/// An item is available when it is cached for browsing (`synced_at`, only +/// while the catalog-browse flag is on — offline with the toggle off the page +/// shows downloaded media only; see `set_include_catalog_browse`), when it is +/// a playable item with a completed download, or when it is a container with a +/// downloaded child. That is checked per row with `EXISTS` on the rows the +/// parent selects. It used to be a CTE that materialised the id of *every* +/// available item in the database (a `UNION` over the whole table) before +/// filtering to the parent, which cost ~80 ms on a desktop for a 100k-item +/// cache whatever the parent — several times that on a phone. +/// +/// A parent that is not a library is matched on the hierarchy columns alone, +/// which SQLite serves with one index lookup per column. The library clause is +/// dropped there rather than evaluated: it can only match when the parent *is* +/// a library, and inside the same `OR` it forced a scan of every row. The +/// `+` on `server_id` stops the planner — which has no statistics, the app +/// never runs `ANALYZE` — from choosing the server index, which every row +/// shares. +/// +/// Bind order: server id; the parent id four times (plus a fifth for a library +/// parent); the type-filter values; with the favourites filter, the user id. +/// +/// TRACES: UR-002, UR-007 | DR-013, DR-277 +#[allow(clippy::too_many_arguments)] +fn items_listing_sql( + parent_is_library: bool, + include_catalog: bool, + type_filter: &str, + favorites_filter: &str, + order_by: &str, + limit: usize, + start_index: usize, +) -> String { + let catalog_available = if include_catalog { + "i.synced_at IS NOT NULL OR " + } else { + "" + }; + let parent_match = if parent_is_library { + format!( + "i.server_id = ? + AND ( + i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ? + -- When the requested parent is a LIBRARY, there is no per-item + -- link back to it (library_id/parent_id are NULL in the cache), + -- so match every item on the server and let the type filter + -- (e.g. MusicAlbum / Movie / Series) narrow it. This is what + -- makes library landing pages show albums/movies/shows offline. + -- + -- The type correlation is NOT optional. Without it this + -- EXISTS never mentions the item, so it is true for every + -- cached row as soon as the requested parent is any library. + -- Music/Movies/TV got away with that because their landing + -- pages pass `include_item_types`, which narrowed the result; + -- the generic library page passes none, so a Books or Photos + -- library served the entire cached server (DR-277). + -- + -- `library_id` wins wherever it survived the cache write: + -- it is the server's own answer, and it is the only thing + -- that can scope a library whose type has no mapping (Books, + -- Photos, Collections) or none at all (a mixed library, where + -- Jellyfin sends CollectionType null). The taxonomy is the + -- fallback for rows that predate it being stored. + -- + -- A library with neither a stored link nor a mapped type now + -- matches nothing here and falls through to the server, which + -- does know what is in it. Showing nothing briefly beats + -- showing somebody else's films with confidence. + OR EXISTS ( + SELECT 1 FROM libraries l + WHERE l.id = ? AND l.server_id = i.server_id + AND ( + i.library_id = l.id + OR (i.library_id IS NULL AND {}) + ) + ) + )", + library_type_matches_item!() + ) + } else { + "+i.server_id = ? + AND (i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ?)" + .to_string() + }; + format!( + "SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres, + i.runtime_ticks, i.production_year, i.community_rating, i.official_rating, + i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists, + i.index_number, i.series_id, i.series_name, i.season_id, i.season_name, + i.parent_index_number, i.is_folder, i.premiere_date + FROM items i + WHERE {parent_match} + AND ( + {catalog_available}( + i.item_type IN ('Audio', 'Movie', 'Episode') + AND EXISTS ( + SELECT 1 FROM downloads d + WHERE d.item_id = i.id AND d.status = 'completed' + ) + ) + OR ( + i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder') + AND EXISTS ( + SELECT 1 FROM items children + INNER JOIN downloads d ON d.item_id = children.id AND d.status = 'completed' + WHERE children.parent_id = i.id OR children.album_id = i.id + OR children.season_id = i.id OR children.series_id = i.id + ) + ) + ){type_filter}{favorites_filter} + ORDER BY {order_by} + LIMIT {limit} OFFSET {start_index}" + ) } // Helper struct matching storage.rs CachedItem structure @@ -475,23 +653,99 @@ impl OfflineRepository { let now = chrono::Utc::now().to_rfc3339(); - // Temporarily disable foreign key constraints to avoid CASCADE DELETE issues - // when replacing stub parent items with their actual data + // Which library do these items belong to? + // + // Resolved once per call, from the parent being browsed. Two cases and + // nothing else: + // + // * the parent IS a library -> these are its direct children + // * the parent is an item -> inherit whatever library that item is + // already known to belong to, so tracks + // under an album and episodes under a + // season land in the same library as + // their container + // + // Synthetic parents ("favorites" and friends) match neither and stay + // NULL, which is correct: they are not a library and their contents + // span several. + // + // Until this existed, `library_id` was bound NULL for every cached row + // and the only way to associate an item with a library was the + // `collection_type` ↔ `item_type` taxonomy. That cannot tell two + // libraries of the *same* type apart — a server with "TV" and "Shows" + // served both the same contents — and has nothing to say about a + // library whose type it does not map (DR-278). + // + // TRACES: UR-007 | DR-278 + let owning_library = self.resolve_owning_library(parent_id).await; + + // Stub rows for every parent referenced, so `items.parent_id` resolves + // whatever order the server returned children and parents in. + let mut parent_ids = std::collections::HashSet::new(); + parent_ids.insert(parent_id.to_string()); + for item in items { + if let Some(pid) = &item.parent_id { + parent_ids.insert(pid.clone()); + } + } + let stubs: Vec = parent_ids + .into_iter() + .map(|pid| { + Query::with_params( + "INSERT OR IGNORE INTO items (id, server_id, name, item_type, synced_at) + VALUES (?1, ?2, ?3, ?4, ?5)", + vec![ + QueryParam::String(pid), + QueryParam::String(self.server_id.clone()), + QueryParam::String("Parent".to_string()), + QueryParam::String("Folder".to_string()), + QueryParam::String(now.clone()), + ], + ) + }) + .collect(); + + let rows: Vec<(String, Query, Option)> = items + .iter() + .map(|item| { + ( + item.id.clone(), + self.item_upsert_query(item, &owning_library, &now), + self.user_data_mirror_query(item, &now), + ) + }) + .collect(); + + // The whole page is one transaction — one job on the writer. It used to + // be one commit per stub, item and user_data row: opening a series + // queued hundreds of them, and every read in the app waited behind the + // pile. + // + // Foreign-key checks are off for this job only. They used to be toggled + // on the shared connection around the save's many awaits, so they were + // off for every *other* write that ran in between too. + // + // TRACES: UR-002, UR-007 | DR-012 self.db_service - .execute(Query::new("PRAGMA foreign_keys = OFF")) + .transaction_without_foreign_keys(move |tx| { + for stub in stubs { + tx.execute(stub)?; + } + let mut count = 0; + for (id, item_query, user_data_query) in rows { + tx.execute(item_query) + .map_err(|e| format!("Failed to insert item {}: {}", id, e))?; + if let Some(query) = user_data_query { + if let Err(e) = tx.execute(query) { + debug!("[OfflineRepo] user_data mirror skipped for {}: {}", id, e); + } + } + count += 1; + } + Ok(count) + }) .await - .map_err(|e| RepoError::Database { message: e })?; - - // Ensure we re-enable foreign keys even if an error occurs - let result = self.save_to_cache_impl(parent_id, items, &now).await; - - // Re-enable foreign key constraints - let _ = self - .db_service - .execute(Query::new("PRAGMA foreign_keys = ON")) - .await; - - result + .map_err(|e| RepoError::Database { message: e }) } /// Which library the children of `parent_id` belong to. @@ -537,279 +791,193 @@ impl OfflineRepository { .flatten() } - async fn save_to_cache_impl( + /// The UPSERT that caches one item under `owning_library`. + fn item_upsert_query( &self, - parent_id: &str, - items: &[MediaItem], + item: &MediaItem, + owning_library: &Option, now: &str, - ) -> Result { - // Which library do these items belong to? + ) -> Query { + // Convert Option> to JSON strings for storage + let genres_json = item + .genres + .as_ref() + .map(|g| serde_json::to_string(g).unwrap_or_else(|_| "[]".to_string())); + let artists_json = item + .artists + .as_ref() + .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string())); + let backdrop_tags_json = item + .backdrop_image_tags + .as_ref() + .map(|b| serde_json::to_string(b).unwrap_or_else(|_| "[]".to_string())); + + // A real UPSERT, not INSERT OR REPLACE. REPLACE deletes the existing + // row and inserts a new one, which (a) fires no AFTER DELETE trigger + // unless `recursive_triggers` is on — it is not, so `items_ad` never + // ran and the old `items_fts` row was orphaned — and (b) assigns a + // *fresh rowid*, because `items.id` is a TEXT PRIMARY KEY, so + // `items_ai` then appended a second index entry. The result was one + // duplicate FTS index per catalog pass. ON CONFLICT keeps the rowid + // that `items_fts.content_rowid` refers to and fires `items_au`, + // which correctly replaces the entry. // - // Resolved once per call, from the parent being browsed. Two cases and - // nothing else: + // It also preserves columns absent from this statement (etag, + // sort_name, tagline, …) instead of resetting them to defaults the + // way REPLACE did. // - // * the parent IS a library -> these are its direct children - // * the parent is an item -> inherit whatever library that item is - // already known to belong to, so tracks - // under an album and episodes under a - // season land in the same library as - // their container - // - // Synthetic parents ("favorites" and friends) match neither and stay - // NULL, which is correct: they are not a library and their contents - // span several. - // - // Until this existed, `library_id` was bound NULL for every cached row - // and the only way to associate an item with a library was the - // `collection_type` ↔ `item_type` taxonomy. That cannot tell two - // libraries of the *same* type apart — a server with "TV" and "Shows" - // served both the same contents — and has nothing to say about a - // library whose type it does not map (DR-278). - // - // TRACES: UR-007 | DR-278 - let owning_library = self.resolve_owning_library(parent_id).await; - - // Collect all unique parent IDs referenced by items being saved - let mut parent_ids = std::collections::HashSet::new(); - parent_ids.insert(parent_id.to_string()); - - for item in items { - if let Some(pid) = &item.parent_id { - parent_ids.insert(pid.clone()); - } - } - - // Insert stub entries for all parent IDs to satisfy FK constraints - #[cfg(test)] - println!("Creating stub parents for: {:?}", parent_ids); - - for pid in parent_ids { - let parent_query = Query::with_params( - "INSERT OR IGNORE INTO items (id, server_id, name, item_type, synced_at) - VALUES (?1, ?2, ?3, ?4, ?5)", - vec![ - QueryParam::String(pid.clone()), - QueryParam::String(self.server_id.clone()), - QueryParam::String("Parent".to_string()), - QueryParam::String("Folder".to_string()), - QueryParam::String(now.to_string()), - ], - ); - - let _stub_rows = self - .db_service - .execute(parent_query) - .await - .map_err(|e| RepoError::Database { message: e })?; - #[cfg(test)] - println!( - " Created stub parent {} (rows affected: {})", - pid, _stub_rows - ); - } - - let mut count = 0; - - for item in items { - // Convert Option> to JSON strings for storage - let genres_json = item - .genres - .as_ref() - .map(|g| serde_json::to_string(g).unwrap_or_else(|_| "[]".to_string())); - let artists_json = item - .artists - .as_ref() - .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string())); - let backdrop_tags_json = item - .backdrop_image_tags - .as_ref() - .map(|b| serde_json::to_string(b).unwrap_or_else(|_| "[]".to_string())); - - // A real UPSERT, not INSERT OR REPLACE. REPLACE deletes the existing - // row and inserts a new one, which (a) fires no AFTER DELETE trigger - // unless `recursive_triggers` is on — it is not, so `items_ad` never - // ran and the old `items_fts` row was orphaned — and (b) assigns a - // *fresh rowid*, because `items.id` is a TEXT PRIMARY KEY, so - // `items_ai` then appended a second index entry. The result was one - // duplicate FTS index per catalog pass. ON CONFLICT keeps the rowid - // that `items_fts.content_rowid` refers to and fires `items_au`, - // which correctly replaces the entry. - // - // It also preserves columns absent from this statement (etag, - // sort_name, tagline, …) instead of resetting them to defaults the - // way REPLACE did. - // - // TRACES: UR-065 | DR-110 | UT-112 - let query = Query::with_params( - "INSERT INTO items ( - id, server_id, library_id, parent_id, - name, item_type, is_folder, overview, - genres, series_id, series_name, - season_id, season_name, index_number, parent_index_number, - album_id, album_name, album_artist, artists, - production_year, premiere_date, runtime_ticks, - primary_image_tag, backdrop_image_tags, - community_rating, official_rating, - synced_at - ) VALUES ( - ?1, ?2, ?3, ?4, - ?5, ?6, ?7, ?8, - ?9, ?10, ?11, - ?12, ?13, ?14, ?15, - ?16, ?17, ?18, ?19, - ?20, ?21, ?22, - ?23, ?24, - ?25, ?26, - ?27 - ) - ON CONFLICT(id) DO UPDATE SET - server_id = excluded.server_id, - -- This call site never supplies library_id (it is always - -- bound NULL), so keep whatever another path recorded rather - -- than clearing it the way REPLACE did. - library_id = COALESCE(excluded.library_id, items.library_id), - parent_id = excluded.parent_id, - name = excluded.name, - item_type = excluded.item_type, - is_folder = excluded.is_folder, - overview = excluded.overview, - genres = excluded.genres, - series_id = excluded.series_id, - series_name = excluded.series_name, - season_id = excluded.season_id, - season_name = excluded.season_name, - index_number = excluded.index_number, - parent_index_number = excluded.parent_index_number, - album_id = excluded.album_id, - album_name = excluded.album_name, - album_artist = excluded.album_artist, - artists = excluded.artists, - production_year = excluded.production_year, - premiere_date = excluded.premiere_date, - runtime_ticks = excluded.runtime_ticks, - primary_image_tag = excluded.primary_image_tag, - backdrop_image_tags = excluded.backdrop_image_tags, - community_rating = excluded.community_rating, - official_rating = excluded.official_rating, - synced_at = excluded.synced_at", - vec![ - QueryParam::String(item.id.clone()), - QueryParam::String(self.server_id.clone()), - // The library this browse belongs to; NULL only for - // synthetic parents. See `resolve_owning_library`. - match &owning_library { - Some(lib) => QueryParam::String(lib.clone()), - None => QueryParam::Null, - }, // library_id - // Use the item's actual parent_id, not the function parameter - match &item.parent_id { - Some(pid) => QueryParam::String(pid.clone()), - None => QueryParam::Null, - }, - QueryParam::String(item.name.clone()), - QueryParam::String(item.item_type.clone()), - QueryParam::Int(if item.is_folder { 1 } else { 0 }), - match &item.overview { - Some(o) => QueryParam::String(o.clone()), - None => QueryParam::Null, - }, - match genres_json { - Some(g) => QueryParam::String(g), - None => QueryParam::Null, - }, - match &item.series_id { - Some(s) => QueryParam::String(s.clone()), - None => QueryParam::Null, - }, - match &item.series_name { - Some(s) => QueryParam::String(s.clone()), - None => QueryParam::Null, - }, - match &item.season_id { - Some(s) => QueryParam::String(s.clone()), - None => QueryParam::Null, - }, - match &item.season_name { - Some(s) => QueryParam::String(s.clone()), - None => QueryParam::Null, - }, - match item.index_number { - Some(i) => QueryParam::Int(i), - None => QueryParam::Null, - }, - match item.parent_index_number { - Some(i) => QueryParam::Int(i), - None => QueryParam::Null, - }, - match &item.album_id { - Some(a) => QueryParam::String(a.clone()), - None => QueryParam::Null, - }, - match &item.album_name { - Some(a) => QueryParam::String(a.clone()), - None => QueryParam::Null, - }, - match &item.album_artist { - Some(a) => QueryParam::String(a.clone()), - None => QueryParam::Null, - }, - match artists_json { - Some(a) => QueryParam::String(a), - None => QueryParam::Null, - }, - match item.production_year { - Some(y) => QueryParam::Int(y), - None => QueryParam::Null, - }, - match &item.premiere_date { - Some(d) => QueryParam::String(d.clone()), - None => QueryParam::Null, - }, - match item.runtime_ticks { - Some(r) => QueryParam::Int64(r), - None => QueryParam::Null, - }, - match &item.primary_image_tag { - Some(t) => QueryParam::String(t.clone()), - None => QueryParam::Null, - }, - match backdrop_tags_json { - Some(b) => QueryParam::String(b), - None => QueryParam::Null, - }, - match item.community_rating { - Some(r) => QueryParam::Float(r), - None => QueryParam::Null, - }, - match &item.official_rating { - Some(r) => QueryParam::String(r.clone()), - None => QueryParam::Null, - }, - QueryParam::String(now.to_string()), - ], - ); - - let _rows_affected = - self.db_service - .execute(query) - .await - .map_err(|e| RepoError::Database { - message: format!("Failed to insert item {}: {}", item.id, e), - })?; - #[cfg(test)] - println!( - " [save_to_cache] Saved item {} (rows affected: {})", - item.id, _rows_affected - ); - - self.mirror_user_data(item, now).await?; - count += 1; - } - - Ok(count) + // TRACES: UR-065 | DR-110 | UT-112 + Query::with_params( + "INSERT INTO items ( + id, server_id, library_id, parent_id, + name, item_type, is_folder, overview, + genres, series_id, series_name, + season_id, season_name, index_number, parent_index_number, + album_id, album_name, album_artist, artists, + production_year, premiere_date, runtime_ticks, + primary_image_tag, backdrop_image_tags, + community_rating, official_rating, + synced_at + ) VALUES ( + ?1, ?2, ?3, ?4, + ?5, ?6, ?7, ?8, + ?9, ?10, ?11, + ?12, ?13, ?14, ?15, + ?16, ?17, ?18, ?19, + ?20, ?21, ?22, + ?23, ?24, + ?25, ?26, + ?27 + ) + ON CONFLICT(id) DO UPDATE SET + server_id = excluded.server_id, + -- This call site never supplies library_id (it is always + -- bound NULL), so keep whatever another path recorded rather + -- than clearing it the way REPLACE did. + library_id = COALESCE(excluded.library_id, items.library_id), + parent_id = excluded.parent_id, + name = excluded.name, + item_type = excluded.item_type, + is_folder = excluded.is_folder, + overview = excluded.overview, + genres = excluded.genres, + series_id = excluded.series_id, + series_name = excluded.series_name, + season_id = excluded.season_id, + season_name = excluded.season_name, + index_number = excluded.index_number, + parent_index_number = excluded.parent_index_number, + album_id = excluded.album_id, + album_name = excluded.album_name, + album_artist = excluded.album_artist, + artists = excluded.artists, + production_year = excluded.production_year, + premiere_date = excluded.premiere_date, + runtime_ticks = excluded.runtime_ticks, + primary_image_tag = excluded.primary_image_tag, + backdrop_image_tags = excluded.backdrop_image_tags, + community_rating = excluded.community_rating, + official_rating = excluded.official_rating, + synced_at = excluded.synced_at", + vec![ + QueryParam::String(item.id.clone()), + QueryParam::String(self.server_id.clone()), + // The library this browse belongs to; NULL only for + // synthetic parents. See `resolve_owning_library`. + match &owning_library { + Some(lib) => QueryParam::String(lib.clone()), + None => QueryParam::Null, + }, // library_id + // Use the item's actual parent_id, not the function parameter + match &item.parent_id { + Some(pid) => QueryParam::String(pid.clone()), + None => QueryParam::Null, + }, + QueryParam::String(item.name.clone()), + QueryParam::String(item.item_type.clone()), + QueryParam::Int(if item.is_folder { 1 } else { 0 }), + match &item.overview { + Some(o) => QueryParam::String(o.clone()), + None => QueryParam::Null, + }, + match genres_json { + Some(g) => QueryParam::String(g), + None => QueryParam::Null, + }, + match &item.series_id { + Some(s) => QueryParam::String(s.clone()), + None => QueryParam::Null, + }, + match &item.series_name { + Some(s) => QueryParam::String(s.clone()), + None => QueryParam::Null, + }, + match &item.season_id { + Some(s) => QueryParam::String(s.clone()), + None => QueryParam::Null, + }, + match &item.season_name { + Some(s) => QueryParam::String(s.clone()), + None => QueryParam::Null, + }, + match item.index_number { + Some(i) => QueryParam::Int(i), + None => QueryParam::Null, + }, + match item.parent_index_number { + Some(i) => QueryParam::Int(i), + None => QueryParam::Null, + }, + match &item.album_id { + Some(a) => QueryParam::String(a.clone()), + None => QueryParam::Null, + }, + match &item.album_name { + Some(a) => QueryParam::String(a.clone()), + None => QueryParam::Null, + }, + match &item.album_artist { + Some(a) => QueryParam::String(a.clone()), + None => QueryParam::Null, + }, + match artists_json { + Some(a) => QueryParam::String(a), + None => QueryParam::Null, + }, + match item.production_year { + Some(y) => QueryParam::Int(y), + None => QueryParam::Null, + }, + match &item.premiere_date { + Some(d) => QueryParam::String(d.clone()), + None => QueryParam::Null, + }, + match item.runtime_ticks { + Some(r) => QueryParam::Int64(r), + None => QueryParam::Null, + }, + match &item.primary_image_tag { + Some(t) => QueryParam::String(t.clone()), + None => QueryParam::Null, + }, + match backdrop_tags_json { + Some(b) => QueryParam::String(b), + None => QueryParam::Null, + }, + match item.community_rating { + Some(r) => QueryParam::Float(r), + None => QueryParam::Null, + }, + match &item.official_rating { + Some(r) => QueryParam::String(r.clone()), + None => QueryParam::Null, + }, + QueryParam::String(now.to_string()), + ], + ) } - /// Mirror the server's per-user state for an item into the local + /// The write that mirrors the server's per-user state for an item into the local /// `user_data` table, so favourites marked — and positions watched — on any /// other client are visible here, including offline, where the local table /// is the only source. @@ -834,8 +1002,11 @@ impl OfflineRepository { /// episode back as unwatched — the list the season view ticks and the one /// `pick_current_episode` reads to decide what is up next (DR-264). /// + /// Best-effort: `save_to_cache` logs and skips a failure rather than + /// failing the whole cache write over it, which would break browsing. + /// /// TRACES: UR-025, UR-062, UR-069 | DR-114, DR-155, DR-264 | UT-102, UT-152, UT-240 - async fn mirror_user_data(&self, item: &MediaItem, now: &str) -> Result<(), RepoError> { + fn user_data_mirror_query(&self, item: &MediaItem, now: &str) -> Option { let user_data = item.user_data.as_ref(); let is_favorite = user_data.and_then(|ud| ud.is_favorite); let position_ticks = user_data.and_then(|ud| ud.playback_position_ticks); @@ -843,7 +1014,7 @@ impl OfflineRepository { // Nothing the server actually told us about — do not invent a row. if is_favorite.is_none() && position_ticks.is_none() && is_played.is_none() { - return Ok(()); + return None; } let query = Query::with_params( @@ -874,17 +1045,7 @@ impl OfflineRepository { ], ); - // A missing item row (FK) is not fatal here — the mirror is best-effort - // metadata, and failing the whole cache write over it would break - // browsing. - if let Err(e) = self.db_service.execute(query).await { - debug!( - "[OfflineRepo] user_data mirror skipped for {}: {}", - item.id, e - ); - } - - Ok(()) + Some(query) } /// Cache the library (view) list from the server into the local database. @@ -1125,11 +1286,7 @@ impl OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; let total_record_count = items.len(); Ok(SearchResult { @@ -1450,96 +1607,32 @@ impl MediaRepository for OfflineRepository { "" }; - // Use CTE to find items that are either: - // 1. Playable items (Audio, Movie, Episode) with completed downloads (offline mode) - // 2. Container items (MusicAlbum, Series, Season) with at least one downloaded child (offline mode) - // 3. Cached items with recent synced_at timestamp (fast online browsing, or the - // offline "Show all server media" catalog view) — only when the catalog-browse - // flag is set. When offline with the toggle off, this branch is omitted so the - // page shows downloaded/local media only. See `set_include_catalog_browse`. - let catalog_branch = if include_catalog_browse() { - "UNION - - -- Cached items for fast browsing (online) or the offline catalog view - SELECT DISTINCT i.id - FROM items i - WHERE i.synced_at IS NOT NULL" - } else { - "" - }; - let sql = format!( - "WITH available_items AS ( - -- Playable items with completed downloads - SELECT DISTINCT i.id - FROM items i - INNER JOIN downloads d ON i.id = d.item_id - WHERE d.status = 'completed' - AND i.item_type IN ('Audio', 'Movie', 'Episode') - - UNION - - -- Containers with downloaded children - SELECT DISTINCT i.id - FROM items i - INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id) - INNER JOIN downloads d ON children.id = d.item_id - WHERE d.status = 'completed' - AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder') - - {catalog_branch} + // Decided up front so the listing can use the hierarchy indexes; see + // `items_listing_sql`. + let parent_is_library = self + .db_service + .query_optional( + Query::with_params( + "SELECT 1 FROM libraries WHERE id = ? AND server_id = ?", + vec![ + QueryParam::String(parent_id.to_string()), + QueryParam::String(self.server_id.clone()), + ], + ), + |row| row.get::<_, i64>(0), ) - SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres, - i.runtime_ticks, i.production_year, i.community_rating, i.official_rating, - i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists, - i.index_number, i.series_id, i.series_name, i.season_id, i.season_name, - i.parent_index_number, i.is_folder, i.premiere_date - FROM items i - INNER JOIN available_items ai ON i.id = ai.id - WHERE i.server_id = ? - AND ( - i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ? - -- When the requested parent is a LIBRARY, there is no per-item - -- link back to it (library_id/parent_id are NULL in the cache), - -- so match every item on the server and let the type filter - -- (e.g. MusicAlbum / Movie / Series) narrow it. This is what - -- makes library landing pages show albums/movies/shows offline. - -- - -- The type correlation is NOT optional. Without it this - -- EXISTS never mentions the item, so it is true for every - -- cached row as soon as the requested parent is any library. - -- Music/Movies/TV got away with that because their landing - -- pages pass `include_item_types`, which narrowed the result; - -- the generic library page passes none, so a Books or Photos - -- library served the entire cached server (DR-277). - -- - -- `library_id` wins wherever it survived the cache write: - -- it is the server's own answer, and it is the only thing - -- that can scope a library whose type has no mapping (Books, - -- Photos, Collections) or none at all (a mixed library, where - -- Jellyfin sends CollectionType null). The taxonomy is the - -- fallback for rows that predate it being stored. - -- - -- A library with neither a stored link nor a mapped type now - -- matches nothing here and falls through to the server, which - -- does know what is in it. Showing nothing briefly beats - -- showing somebody else's films with confidence. - OR EXISTS ( - SELECT 1 FROM libraries l - WHERE l.id = ? AND l.server_id = i.server_id - AND ( - i.library_id = l.id - OR (i.library_id IS NULL AND {}) - ) - ) - ){}{} - ORDER BY {} - LIMIT {} OFFSET {}", - library_type_matches_item!(), - type_filter, + .await + .map_err(|e| RepoError::Database { message: e })? + .is_some(); + + let sql = items_listing_sql( + parent_is_library, + include_catalog_browse(), + &type_filter, favorites_filter, - order_by, + &order_by, limit, - start_index + start_index, ); // The requested id is compared against every hierarchy-linkage column @@ -1553,12 +1646,14 @@ impl MediaRepository for OfflineRepository { QueryParam::String(parent_id.to_string()), // i.album_id = ? QueryParam::String(parent_id.to_string()), // i.season_id = ? QueryParam::String(parent_id.to_string()), // i.series_id = ? - QueryParam::String(parent_id.to_string()), // libraries.id = ? ]; + if parent_is_library { + params.push(QueryParam::String(parent_id.to_string())); // libraries.id = ? + } // Positional order matters: the type placeholders sit in `{type_filter}`, // which the statement interpolates immediately after the parent-matching // group and before `{favorites_filter}`, so they bind here — after the - // six ids above, before the favourites user id. + // ids above, before the favourites user id. params.extend(type_values.iter().cloned().map(QueryParam::String)); if !favorites_filter.is_empty() { params.push(QueryParam::String(self.user_id.clone())); // ud.user_id = ? @@ -1577,12 +1672,7 @@ impl MediaRepository for OfflineRepository { &parent_id[..8.min(parent_id.len())] ); - // Fetch user data for each item - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; let total_record_count = items.len(); @@ -1704,11 +1794,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; Ok(items) } @@ -1782,11 +1868,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; Ok(items) } @@ -1872,11 +1954,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; Ok(items) } @@ -1926,11 +2004,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; Ok(items) } @@ -2064,11 +2138,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let mut items = self.with_user_data(cached_items).await; // People are not rows in `items` — they live in their own table — so // they need a second lookup. Only when the scope admits them: an @@ -2293,11 +2363,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; let total_record_count = items.len(); debug!("[OfflineRepo] Returning {} favourites", total_record_count); @@ -2436,11 +2502,7 @@ impl MediaRepository for OfflineRepository { .await .map_err(|e| RepoError::Database { message: e })?; - let mut items = Vec::new(); - for cached in cached_items { - let user_data = self.get_user_data(&cached.id).await; - items.push(Self::cached_item_to_media_item(cached, user_data)); - } + let items = self.with_user_data(cached_items).await; let total_record_count = items.len(); @@ -3106,9 +3168,9 @@ mod tests { /// /// `get_items` gave the cache 100 ms and *discarded* a slower answer, then /// waited on the server — which offline fails — and returned the server's - /// error. The cache is one SQLite connection behind one mutex, so any write - /// in progress (the catalog sync that starts at every launch, a download - /// finishing) pushes a read past 100 ms routinely. Every other cached query + /// error. The cache was then one SQLite connection behind one mutex, so any + /// write in progress (the catalog sync that starts at every launch, a + /// download finishing) pushed a read past 100 ms routinely. Every other cached query /// already keeps a slow read alive and waits for it when the server fails; /// `get_items`, which the series page calls for the show and each season, /// did not. @@ -5612,4 +5674,164 @@ mod tests { ); } } + + /// Caching a page must not switch off foreign-key enforcement for anyone + /// else. + /// + /// `save_to_cache` turns FK checks off while it writes stub parents. That + /// pragma is per *connection*, and the connection is shared, so it used to + /// stay off across every await of the save — any other write that ran in + /// that window (an orphan row, a cascade that should have fired) went + /// through unchecked. The toggle must be confined to the cache write alone. + /// + /// TRACES: UR-002 | DR-012 | UT-014 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn save_to_cache_does_not_disable_foreign_keys_for_concurrent_writes() { + // The fixture already has the `test-server` row, so the only thing an + // orphan below can violate is its parent reference. + let db_service = create_test_db(); + + let repo = OfflineRepository::new( + db_service.clone(), + "test-server".to_string(), + "test-user".to_string(), + ); + let items: Vec = (0..1500) + .map(|i| create_test_item(&format!("item-{i}"), "Item", Some("library-1"))) + .collect(); + + let save = tokio::spawn(async move { repo.save_to_cache("library-1", &items).await }); + + // Hammer the connection with FK-violating writes while the save runs. + let mut orphans_accepted = 0; + let mut attempt = 0; + while !save.is_finished() { + let inserted = db_service + .execute(Query::with_params( + "INSERT INTO items (id, server_id, name, item_type, parent_id) + VALUES (?, 'test-server', 'Orphan', 'Audio', 'no-such-parent')", + vec![QueryParam::String(format!("orphan-{attempt}"))], + )) + .await; + if inserted.is_ok() { + orphans_accepted += 1; + } + attempt += 1; + } + save.await.unwrap().unwrap(); + + assert!( + attempt > 0, + "the save finished before any write could race it" + ); + assert_eq!( + orphans_accepted, 0, + "{orphans_accepted} of {attempt} FK-violating writes were accepted mid-save" + ); + } + + /// Listings attach user data in batches, and each row must still get its + /// *own* — across the chunk boundary, and `None` for rows without any. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn listing_user_data_is_batched_per_row() { + let _guard = lock_catalog_browse(); + set_include_catalog_browse(true); + let db_service = create_test_db(); + let repo = OfflineRepository::new( + db_service.clone(), + "test-server".to_string(), + "test-user".to_string(), + ); + + // More rows than one batch; every third has a position of its index. + let items: Vec = (0..1203) + .map(|i| { + create_test_item( + &format!("ep-{i:04}"), + &format!("Ep {i:04}"), + Some("season-1"), + ) + }) + .collect(); + repo.save_to_cache("season-1", &items).await.unwrap(); + for i in (0..1203).step_by(3) { + db_service + .execute(Query::with_params( + "INSERT INTO user_data (user_id, item_id, playback_position_ticks) VALUES ('test-user', ?, ?)", + vec![QueryParam::String(format!("ep-{i:04}")), QueryParam::Int64(i as i64)], + )) + .await + .unwrap(); + } + + let result = repo.get_items("season-1", None).await.unwrap(); + assert_eq!(result.items.len(), 1203); + for item in &result.items { + let i: i64 = item.id.trim_start_matches("ep-").parse().unwrap(); + let position = item + .user_data + .as_ref() + .and_then(|u| u.playback_position_ticks); + if i % 3 == 0 { + assert_eq!( + position, + Some(i), + "{} got someone else's user data", + item.id + ); + } else { + assert_eq!(position, None, "{} got user data it does not have", item.id); + } + } + } + + /// Listing a series, season or album must look up its children by index, + /// not walk the whole catalogue. + /// + /// The query used to build the set of *every* available item id (a + /// `UNION` over the whole `items` table) and then filter it to the parent, + /// and its library clause sat in the same `OR` as the parent columns, which + /// forced a scan of every row on the server. About 80 ms on a desktop for a + /// 100k-item cache — several times that on a phone, so the series page's + /// read missed the cache fast path on every visit. The app never runs + /// `ANALYZE`, so the plan must be good without statistics. + /// + /// TRACES: UR-002 | DR-013 + #[test] + fn listing_a_non_library_parent_uses_the_hierarchy_indexes() { + use crate::utils::lock::MutexSafe; + let db = crate::storage::Database::open_in_memory().unwrap(); + let conn = db.connection(); + let conn = conn.lock_safe(); + for include_catalog in [true, false] { + let sql = items_listing_sql( + false, + include_catalog, + "", + "", + "i.sort_name ASC, i.name ASC", + 100, + 0, + ); + let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap(); + let plan: Vec = stmt + .query_map(rusqlite::params!["srv", "p", "p", "p", "p"], |row| { + row.get::<_, String>(3) + }) + .unwrap() + .map(Result::unwrap) + .collect(); + let plan = plan.join("\n"); + assert!( + plan.contains("MULTI-INDEX OR"), + "expected an index lookup per hierarchy column, got:\n{plan}" + ); + assert!( + !plan.contains("SCAN i") && !plan.contains("idx_items_server"), + "the listing walks every item on the server:\n{plan}" + ); + } + } } diff --git a/src-tauri/src/storage/db_service.rs b/src-tauri/src/storage/db_service.rs index 8fdf96799..78d5b623f 100644 --- a/src-tauri/src/storage/db_service.rs +++ b/src-tauri/src/storage/db_service.rs @@ -1,16 +1,31 @@ -//! Database service abstraction layer +//! Database service: the single owner of the SQLite database. //! -//! This module provides an async database interface that abstracts away -//! the underlying database implementation. This makes it easy to: -//! - Switch between sync (rusqlite) and async (tokio-rusqlite) implementations -//! - Prevent blocking the async runtime with synchronous database calls -//! - Test with different database backends -//! - Migrate to other database systems in the future +//! Every query in the app goes through [`RusqliteService`], which owns the +//! connections and hands work to them — callers never touch a `Connection`. +//! +//! - **Writes** (`execute`, `insert`, `transaction`, …) are sent as jobs to one +//! dedicated writer thread that owns the read-write connection. SQLite allows +//! one writer at a time anyway; owning it on one thread makes that explicit, +//! keeps connection-wide state (pragmas) out of reach of concurrent callers, +//! and parks no tokio blocking threads on a mutex while writes queue up. +//! - **Reads** (`query_*`) run on a small pool of read-only connections. The +//! database is in WAL mode, so readers see the last committed state and never +//! wait for the writer — a large catalog-cache transaction no longer stalls +//! library pages, thumbnail lookups or settings reads. +//! +//! A service built with [`RusqliteService::new`] has no reader pool (in-memory +//! databases cannot be shared between connections) and routes reads through +//! the writer, which is the old single-connection behaviour tests rely on. +//! +//! See `docs/architecture/08-database-design.md` → "Connection ownership". use crate::utils::lock::MutexSafe; use async_trait::async_trait; +use log::{debug, error}; use rusqlite::{params_from_iter, Connection, Result as SqliteResult, Row}; -use std::sync::{Arc, Mutex}; +use std::panic::AssertUnwindSafe; +use std::sync::mpsc; +use std::sync::{Arc, Condvar, Mutex}; /// Database query result type pub type DbResult = Result; @@ -84,8 +99,28 @@ pub trait DatabaseService: Send + Sync { F: FnOnce(&mut Transaction) -> DbResult + Send + 'static, T: Send + 'static; - /// Get the row ID of the most recent successful INSERT - async fn last_insert_rowid(&self) -> DbResult; + /// Run a transaction with foreign-key enforcement switched off for its + /// duration only. + /// + /// `PRAGMA foreign_keys` is per connection and is a no-op inside a + /// transaction, so it has to be flipped around the `BEGIN`/`COMMIT` — and + /// all of that must happen as one job on the writer, or any other write + /// that got in between would run unchecked too. + async fn transaction_without_foreign_keys(&self, f: F) -> DbResult + where + F: FnOnce(&mut Transaction) -> DbResult + Send + 'static, + T: Send + 'static; + + /// Execute an INSERT and return the rowid of the row it inserted. + /// + /// The rowid is read in the same job as the insert. Reading it with a + /// second call would race every other write, returning someone else's id. + async fn insert(&self, query: Query) -> DbResult; + + /// Queue a write without waiting for it — for best-effort bookkeeping (an + /// LRU access time) that must not hold up the caller. It still runs in + /// order with every other write; failures are only logged. + fn execute_detached(&self, query: Query); } /// Transaction handle for batching multiple operations @@ -110,48 +145,183 @@ impl<'a> Transaction<'a> { } } -/// Rusqlite-based database service implementation -/// -/// This implementation wraps synchronous rusqlite operations in tokio::task::spawn_blocking -/// to prevent blocking the async runtime. +type Job = Box; + +/// The thread that owns the read-write connection. Jobs run one at a time, in +/// the order they were sent; the thread exits when the last service handle +/// (and so the last sender) is dropped. +struct Writer { + jobs: mpsc::Sender, +} + +impl Writer { + fn spawn(conn: Arc>) -> Self { + let (jobs, queue) = mpsc::channel::(); + std::thread::Builder::new() + .name("db-writer".into()) + .spawn(move || { + for job in queue { + // The connection stays behind a mutex only so migrations and + // tests can reach it; in the app this thread is its sole user. + // `lock_safe` so a poisoned lock is recovered, not fatal. + let conn = conn.lock_safe(); + // A panicking row mapper must not take the owner down with + // it: the job's reply channel drops, its caller gets an + // error, and the next job runs normally. The guard lives + // outside the unwind, so the mutex is not poisoned either. + if std::panic::catch_unwind(AssertUnwindSafe(|| job(&conn))).is_err() { + error!("[db] a database job panicked; the writer carries on"); + // Undo whatever connection state the job was midway + // through: an open transaction would make the next + // job's BEGIN fail, and a job that switched foreign + // keys off would leave them off for everyone. + if !conn.is_autocommit() { + let _ = conn.execute_batch("ROLLBACK"); + } + let _ = conn.execute_batch("PRAGMA foreign_keys = ON"); + } + } + }) + .expect("failed to spawn the database writer thread"); + Self { jobs } + } + + async fn run(&self, f: F) -> DbResult + where + T: Send + 'static, + F: FnOnce(&Connection) -> DbResult + Send + 'static, + { + let (reply, result) = tokio::sync::oneshot::channel(); + self.jobs + .send(Box::new(move |conn| { + let _ = reply.send(f(conn)); + })) + .map_err(|_| "database writer has stopped".to_string())?; + result + .await + .map_err(|_| "database job panicked".to_string())? + } + + fn run_detached(&self, f: impl FnOnce(&Connection) + Send + 'static) { + if self.jobs.send(Box::new(f)).is_err() { + debug!("[db] writer stopped; dropped a detached write"); + } + } +} + +/// Read-only connections, checked out one per query. WAL gives each a +/// snapshot of the last commit, so they never wait for the writer. +struct ReaderPool { + idle: Mutex>, + returned: Condvar, +} + +impl ReaderPool { + /// Blocking: waits for a free connection. Call from `spawn_blocking`. + fn run(&self, f: impl FnOnce(&Connection) -> T) -> T { + let conn = { + let mut idle = self.idle.lock_safe(); + loop { + if let Some(conn) = idle.pop() { + break conn; + } + idle = self + .returned + .wait(idle) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + } + }; + // Returned on drop, so a panicking mapper does not leak the connection. + let checkout = Checkout { + pool: self, + conn: Some(conn), + }; + f(checkout.conn.as_ref().expect("checked-out connection")) + } +} + +struct Checkout<'a> { + pool: &'a ReaderPool, + conn: Option, +} + +impl Drop for Checkout<'_> { + fn drop(&mut self) { + if let Some(conn) = self.conn.take() { + self.pool.idle.lock_safe().push(conn); + self.pool.returned.notify_one(); + } + } +} + +/// Rusqlite-based database service: a cheap, cloneable handle to the writer +/// thread and reader pool. See the module docs. +#[derive(Clone)] pub struct RusqliteService { - conn: Arc>, + writer: Arc, + readers: Option>, } impl RusqliteService { + /// A service over a single connection: writes *and* reads go through the + /// writer thread. Used for in-memory databases, which cannot be shared + /// between connections. + #[cfg_attr(not(test), allow(dead_code))] pub fn new(conn: Arc>) -> Self { - Self { conn } + Self { + writer: Arc::new(Writer::spawn(conn)), + readers: None, + } + } + + /// A service whose reads run on `readers` — read-only connections to the + /// same (file-backed, WAL-mode) database — alongside the writer. + pub fn with_readers(conn: Arc>, readers: Vec) -> Self { + let readers = (!readers.is_empty()).then(|| { + Arc::new(ReaderPool { + idle: Mutex::new(readers), + returned: Condvar::new(), + }) + }); + Self { + writer: Arc::new(Writer::spawn(conn)), + readers, + } + } + + async fn read(&self, f: F) -> DbResult + where + T: Send + 'static, + F: FnOnce(&Connection) -> DbResult + Send + 'static, + { + match &self.readers { + Some(pool) => { + let pool = Arc::clone(pool); + tokio::task::spawn_blocking(move || pool.run(f)) + .await + .map_err(|e| format!("Task join error: {}", e))? + } + None => self.writer.run(f).await, + } } } #[async_trait] impl DatabaseService for RusqliteService { async fn execute(&self, query: Query) -> DbResult { - let conn = Arc::clone(&self.conn); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - execute_query(&conn, query) - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + self.writer + .run(move |conn| execute_query(conn, query)) + .await } async fn execute_batch(&self, sql: &str) -> DbResult<()> { - let conn = Arc::clone(&self.conn); let sql = sql.to_string(); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - conn.execute_batch(&sql) - .map_err(|e| format!("Execute batch failed: {}", e)) - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + self.writer + .run(move |conn| { + conn.execute_batch(&sql) + .map_err(|e| format!("Execute batch failed: {}", e)) + }) + .await } async fn query_one(&self, query: Query, mapper: F) -> DbResult @@ -159,16 +329,7 @@ impl DatabaseService for RusqliteService { T: Send + 'static, F: Fn(&Row) -> SqliteResult + Send + 'static, { - let conn = Arc::clone(&self.conn); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - query_one(&conn, query, mapper) - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + self.read(move |conn| query_one(conn, query, mapper)).await } async fn query_optional(&self, query: Query, mapper: F) -> DbResult> @@ -176,16 +337,8 @@ impl DatabaseService for RusqliteService { T: Send + 'static, F: Fn(&Row) -> SqliteResult + Send + 'static, { - let conn = Arc::clone(&self.conn); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - query_optional(&conn, query, mapper) - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + self.read(move |conn| query_optional(conn, query, mapper)) + .await } async fn query_many(&self, query: Query, mapper: F) -> DbResult> @@ -193,16 +346,7 @@ impl DatabaseService for RusqliteService { T: Send + 'static, F: Fn(&Row) -> SqliteResult + Send + 'static, { - let conn = Arc::clone(&self.conn); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - query_many(&conn, query, mapper) - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + self.read(move |conn| query_many(conn, query, mapper)).await } async fn transaction(&self, f: F) -> DbResult @@ -210,47 +354,65 @@ impl DatabaseService for RusqliteService { F: FnOnce(&mut Transaction) -> DbResult + Send + 'static, T: Send + 'static, { - let conn = Arc::clone(&self.conn); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - - conn.execute("BEGIN TRANSACTION", []) - .map_err(|e| format!("Failed to begin transaction: {}", e))?; - - let mut transaction = Transaction::new(&conn); - let result = f(&mut transaction); - - match result { - Ok(value) => { - conn.execute("COMMIT", []) - .map_err(|e| format!("Failed to commit transaction: {}", e))?; - Ok(value) - } - Err(e) => { - conn.execute("ROLLBACK", []) - .map_err(|e| format!("Failed to rollback transaction: {}", e))?; - Err(e) - } - } - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + self.writer.run(move |conn| run_transaction(conn, f)).await } - async fn last_insert_rowid(&self) -> DbResult { - let conn = Arc::clone(&self.conn); - tokio::task::spawn_blocking(move || { - // `lock_safe`, not `lock`: this is the busiest lock in the app and a - // panic under the guard would otherwise poison it, failing every - // later query with "poisoned lock" until the process restarts. - let conn = conn.lock_safe(); - Ok(conn.last_insert_rowid()) - }) - .await - .map_err(|e| format!("Task join error: {}", e))? + async fn transaction_without_foreign_keys(&self, f: F) -> DbResult + where + F: FnOnce(&mut Transaction) -> DbResult + Send + 'static, + T: Send + 'static, + { + self.writer + .run(move |conn| { + conn.execute_batch("PRAGMA foreign_keys = OFF") + .map_err(|e| format!("Failed to disable foreign keys: {}", e))?; + let result = run_transaction(conn, f); + // Always restored, whatever the transaction did. + if let Err(e) = conn.execute_batch("PRAGMA foreign_keys = ON") { + error!("[db] failed to re-enable foreign keys: {}", e); + } + result + }) + .await + } + + async fn insert(&self, query: Query) -> DbResult { + self.writer + .run(move |conn| { + execute_query(conn, query)?; + Ok(conn.last_insert_rowid()) + }) + .await + } + + fn execute_detached(&self, query: Query) { + self.writer.run_detached(move |conn| { + if let Err(e) = execute_query(conn, query) { + debug!("[db] detached write failed: {}", e); + } + }); + } +} + +fn run_transaction(conn: &Connection, f: F) -> DbResult +where + F: FnOnce(&mut Transaction) -> DbResult, +{ + conn.execute("BEGIN TRANSACTION", []) + .map_err(|e| format!("Failed to begin transaction: {}", e))?; + + let mut transaction = Transaction::new(conn); + match f(&mut transaction) { + Ok(value) => { + conn.execute("COMMIT", []) + .map_err(|e| format!("Failed to commit transaction: {}", e))?; + Ok(value) + } + Err(e) => { + conn.execute("ROLLBACK", []) + .map_err(|e| format!("Failed to rollback transaction: {}", e))?; + Err(e) + } } } diff --git a/src-tauri/src/storage/mod.rs b/src-tauri/src/storage/mod.rs index 57c030149..3dab59eee 100644 --- a/src-tauri/src/storage/mod.rs +++ b/src-tauri/src/storage/mod.rs @@ -17,9 +17,27 @@ use rusqlite::{Connection, Result as SqliteResult}; pub use db_service::{DatabaseService, RusqliteService}; use schema::MIGRATIONS; -/// Database connection wrapper with thread-safe access +/// How long a connection retries a locked database before erroring — covers a +/// reader meeting a WAL checkpoint, or the writer meeting a reader's snapshot. +const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +/// How many read-only connections serve queries alongside the writer. Reads +/// are short; a few cover a library page's parallel fetches plus background +/// work without holding many file handles. +const READER_CONNECTIONS: usize = 3; + +/// The database: opened once at startup and owned for the life of the app. +/// +/// All access goes through [`Database::service`], which hands out clones of one +/// [`RusqliteService`] — a writer thread plus a pool of read-only connections +/// (see `db_service`). Nothing else opens the database file. pub struct Database { + /// The read-write connection. Owned by the service's writer thread; kept + /// here only for migrations (which run before that thread starts taking + /// work) and for tests. + #[cfg_attr(not(test), allow(dead_code))] conn: Arc>, + service: RusqliteService, path: PathBuf, } @@ -36,18 +54,36 @@ impl Database { // Enable foreign keys conn.execute_batch("PRAGMA foreign_keys = ON;")?; - // Enable WAL mode for better concurrent access + // WAL lets the reader connections run alongside the writer. conn.execute_batch("PRAGMA journal_mode = WAL;")?; + // In WAL mode NORMAL is corruption-safe and skips the fsync FULL pays on + // every commit (a power cut can lose the last commits, an app crash + // cannot). On Android flash that fsync dominated every small write. + conn.execute_batch("PRAGMA synchronous = NORMAL;")?; + conn.busy_timeout(BUSY_TIMEOUT)?; - let db = Self { - conn: Arc::new(Mutex::new(conn)), + let conn = Arc::new(Mutex::new(conn)); + Self::migrate_connection(&conn, MIGRATIONS)?; + + // Readers open after migrations, so they only ever see the final schema. + let readers = (0..READER_CONNECTIONS) + .map(|_| Self::open_reader(path)) + .collect::>>()?; + + Ok(Self { + service: RusqliteService::with_readers(Arc::clone(&conn), readers), + conn, path: path.clone(), - }; + }) + } - // Run migrations - db.migrate()?; - - Ok(db) + /// A connection that can only read. `query_only` makes an accidental write + /// routed to the pool fail loudly instead of racing the writer. + fn open_reader(path: &PathBuf) -> SqliteResult { + let conn = Connection::open(path)?; + conn.busy_timeout(BUSY_TIMEOUT)?; + conn.execute_batch("PRAGMA query_only = ON;")?; + Ok(conn) } /// Open an in-memory database (for testing) @@ -58,15 +94,16 @@ impl Database { // Enable foreign keys conn.execute_batch("PRAGMA foreign_keys = ON;")?; - let db = Self { - conn: Arc::new(Mutex::new(conn)), + let conn = Arc::new(Mutex::new(conn)); + Self::migrate_connection(&conn, MIGRATIONS)?; + + // An in-memory database cannot be shared between connections, so this + // one has no reader pool: reads go through the writer. + Ok(Self { + service: RusqliteService::new(Arc::clone(&conn)), + conn, path: PathBuf::from(":memory:"), - }; - - // Run migrations - db.migrate()?; - - Ok(db) + }) } /// Get connection (for testing) @@ -75,11 +112,19 @@ impl Database { Arc::clone(&self.conn) } - /// Run all pending migrations. + /// Re-run all migrations against an open database (tests only; `open` runs + /// them before the service starts). + #[cfg(test)] pub fn migrate(&self) -> SqliteResult<()> { self.migrate_with(MIGRATIONS) } + /// Test seam: inject a failing migration. See [`Self::migrate_connection`]. + #[cfg(test)] + fn migrate_with(&self, migrations: &[(&str, &str)]) -> SqliteResult<()> { + Self::migrate_connection(&self.conn, migrations) + } + /// Apply `migrations` in order, skipping ones `_migrations` already records. /// /// **Each migration is one transaction, and the `_migrations` row is written @@ -96,12 +141,16 @@ impl Database { /// Every migration is pure DDL/DML, which SQLite runs transactionally — a /// `PRAGMA` or `VACUUM` added to one would not roll back and must not be. /// - /// Split out from [`Self::migrate`] so tests can inject a failing migration. + /// Runs on the bare connection, before the writer thread takes it, so + /// tests can also inject a failing migration through `migrate_with`. /// /// TRACES: UR-002 | DR-012 | UT-014 - fn migrate_with(&self, migrations: &[(&str, &str)]) -> SqliteResult<()> { + fn migrate_connection( + conn: &Mutex, + migrations: &[(&str, &str)], + ) -> SqliteResult<()> { info!("Starting database migrations..."); - let conn = self.conn.lock_safe(); + let conn = conn.lock_safe(); // Create migrations table if it doesn't exist debug!("Creating _migrations table if it doesn't exist..."); @@ -160,12 +209,10 @@ impl Database { Ok(()) } - /// Get a database service for async-safe operations - /// - /// This wraps all blocking database operations in spawn_blocking to prevent - /// freezing the async runtime. + /// A handle to the database service. Cheap: every call returns a clone of + /// the same writer thread and reader pool. pub fn service(&self) -> RusqliteService { - RusqliteService::new(Arc::clone(&self.conn)) + self.service.clone() } /// Get the database file path @@ -880,4 +927,197 @@ mod tests { assert_eq!(user_id, "user2"); assert_eq!(username, "recent_user"); } + + /// A read must not queue behind a long write. + /// + /// The database is in WAL mode precisely so readers can run alongside a + /// writer, but every query used to go through one connection behind one + /// mutex — so a big catalog-cache transaction stalled every library page, + /// thumbnail lookup and settings read in the app until it committed. + /// + /// TRACES: UR-002 | DR-012 | UT-014 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn reads_do_not_wait_for_an_in_flight_write() { + use crate::storage::db_service::{DatabaseService, Query}; + use std::time::{Duration, Instant}; + + let dir = tempfile::tempdir().unwrap(); + let db = Database::open(&dir.path().join("jellytau.db")).unwrap(); + let service = db.service(); + + let writer = service.clone(); + let write = tokio::spawn(async move { + writer + .transaction(|_tx| { + std::thread::sleep(Duration::from_millis(600)); + Ok(()) + }) + .await + }); + // Let the write take the connection first. + tokio::time::sleep(Duration::from_millis(100)).await; + + let started = Instant::now(); + let servers: i64 = service + .query_one(Query::new("SELECT COUNT(*) FROM servers"), |r| r.get(0)) + .await + .unwrap(); + let waited = started.elapsed(); + + assert_eq!(servers, 0); + assert!( + waited < Duration::from_millis(250), + "a read waited {waited:?} for an unrelated write to commit" + ); + write.await.unwrap().unwrap(); + } + + /// Commit cost: in WAL mode `synchronous = NORMAL` is corruption-safe and + /// skips the per-commit fsync that FULL (the default) pays — on Android + /// flash that is the dominant cost of every small write. A busy timeout + /// lets the reader connections ride out a checkpoint instead of failing. + /// + /// TRACES: UR-002 | DR-012 | UT-014 + #[test] + fn open_configures_wal_for_interactive_use() { + let dir = tempfile::tempdir().unwrap(); + let db = Database::open(&dir.path().join("jellytau.db")).unwrap(); + let conn = db.connection(); + let conn = conn.lock_safe(); + + let mode: String = conn + .query_row("PRAGMA journal_mode", [], |r| r.get(0)) + .unwrap(); + let synchronous: i64 = conn + .query_row("PRAGMA synchronous", [], |r| r.get(0)) + .unwrap(); + let busy_timeout: i64 = conn + .query_row("PRAGMA busy_timeout", [], |r| r.get(0)) + .unwrap(); + + assert_eq!(mode, "wal"); + assert_eq!(synchronous, 1, "expected synchronous = NORMAL"); + assert!(busy_timeout > 0, "expected a busy timeout"); + } + + /// Writes a phone-sized catalogue to `$JELLYTAU_BENCH_DB` for timing + /// queries with the `sqlite3` CLI. Not a test; run explicitly with + /// `--ignored`. + #[test] + #[ignore] + fn write_bench_database() { + let Ok(path) = std::env::var("JELLYTAU_BENCH_DB") else { + return; + }; + let _ = std::fs::remove_file(&path); + let db = Database::open(&PathBuf::from(&path)).unwrap(); + let conn = db.connection(); + let conn = conn.lock_safe(); + conn.execute_batch( + "BEGIN; + INSERT INTO servers (id, name, url) VALUES ('srv', 'S', 'http://s'); + INSERT INTO users (id, server_id, username) VALUES ('u', 'srv', 'u'); + INSERT INTO libraries (id, server_id, name, collection_type) VALUES ('tv', 'srv', 'TV', 'tvshows'); + INSERT INTO libraries (id, server_id, name, collection_type) VALUES ('music', 'srv', 'Music', 'music');", + ) + .unwrap(); + let now = "2026-09-23T00:00:00Z"; + let mut item = conn + .prepare( + "INSERT INTO items (id, server_id, library_id, parent_id, name, sort_name, item_type, + series_id, season_id, album_id, synced_at) + VALUES (?1, 'srv', ?2, ?3, ?4, ?4, ?5, ?6, ?7, ?8, ?9)", + ) + .unwrap(); + let none: Option = None; + for s in 0..300 { + let series = format!("series-{s}"); + item.execute(rusqlite::params![ + series, + "tv", + none, + format!("Show {s}"), + "Series", + none, + none, + none, + now + ]) + .unwrap(); + for n in 0..11 { + let season = format!("{series}-s{n}"); + item.execute(rusqlite::params![ + season, + "tv", + series, + format!("Season {n}"), + "Season", + series, + none, + none, + now + ]) + .unwrap(); + for e in 0..24 { + let ep = format!("{season}-e{e}"); + item.execute(rusqlite::params![ + ep, + "tv", + season, + format!("A Title {e}"), + "Episode", + series, + season, + none, + now + ]) + .unwrap(); + conn.execute( + "INSERT INTO user_data (user_id, item_id, playback_position_ticks) VALUES ('u', ?1, 5)", + [&ep], + ) + .unwrap(); + } + } + } + for a in 0..2000 { + let album = format!("album-{a}"); + item.execute(rusqlite::params![ + album, + "music", + none, + format!("Album {a}"), + "MusicAlbum", + none, + none, + none, + now + ]) + .unwrap(); + for t in 0..12 { + item.execute(rusqlite::params![ + format!("{album}-t{t}"), + "music", + album, + format!("Track {t}"), + "Audio", + none, + none, + album, + now + ]) + .unwrap(); + } + } + for d in 0..400 { + conn.execute( + "INSERT INTO downloads (item_id, user_id, file_path, status) VALUES (?1, 'u', '/x', 'completed')", + [format!("series-{}-s1-e{}", d % 300, d % 24)], + ) + .unwrap(); + } + drop(item); + // No ANALYZE: the app never runs it, so the planner works without stats. + conn.execute_batch("COMMIT;").unwrap(); + } } diff --git a/src-tauri/src/thumbnail/cache.rs b/src-tauri/src/thumbnail/cache.rs index e170663b6..54a24a655 100644 --- a/src-tauri/src/thumbnail/cache.rs +++ b/src-tauri/src/thumbnail/cache.rs @@ -122,7 +122,7 @@ impl ThumbnailCache { { let path = PathBuf::from(&path_str); if path.exists() { - self.touch(&db, item_id, image_type, Some(tag)).await; + self.touch(&db, item_id, image_type, Some(tag)); return Some(path); } // File gone — drop the stale row and fall through to the tag-agnostic @@ -157,7 +157,7 @@ impl ThumbnailCache { let path_str: String = db.query_optional(any_tag, |row| row.get(0)).await.ok()??; let path = PathBuf::from(&path_str); if path.exists() { - self.touch(&db, item_id, image_type, None).await; + self.touch(&db, item_id, image_type, None); Some(path) } else { None @@ -166,13 +166,7 @@ impl ThumbnailCache { /// Update `last_accessed` for LRU tracking. When `tag` is `Some`, scope to /// that exact row; when `None`, touch every row for the item + type. - async fn touch( - &self, - db: &Arc, - item_id: &str, - image_type: &str, - tag: Option<&str>, - ) { + fn touch(&self, db: &Arc, item_id: &str, image_type: &str, tag: Option<&str>) { let query = match tag { Some(tag) => Query::with_params( "UPDATE thumbnails SET last_accessed = CURRENT_TIMESTAMP @@ -192,7 +186,10 @@ impl ThumbnailCache { ], ), }; - let _ = db.execute(query).await; + // Detached: an LRU timestamp is bookkeeping, and a grid scroll does + // one of these per visible poster — awaiting each write held every + // thumbnail lookup behind the writer queue. + db.execute_detached(query); } /// Save thumbnail to cache