Compare commits

..
7 Commits
Author SHA1 Message Date
dtourolle 73fd8a1dfe chore(release): v0.13.2
🏗️ Build and Test JellyTau / Run Tests (push) Skipped
🏗️ Build and Test JellyTau / Android Compile Check (push) Skipped
🏗️ Build and Test JellyTau / Supply Chain (push) Successful in 4m15s
📱 Test APK / Build test APK (push) Successful in 20m30s
Publish Documentation / Build & publish docs to gitea-pages (push) Successful in 3m46s
Traceability Validation / Check Requirement Traces (push) Successful in 14s
Build & Release / Run Tests (push) Successful in 10m32s
Build & Release / Build Linux (push) Successful in 14m6s
Build & Release / Build Windows (push) Successful in 10m50s
Build & Release / Build Android (push) Successful in 20m20s
Build & Release / Create Release (push) Successful in 37s
A series opens with its episodes in under a second: the episode list no
longer waits on the server, a page loads once instead of six times, and
the local database finds a container's children by index.
2026-09-24 04:47:07 +02:00
dtourolle a676f4aba8 perf(series): the episode list no longer waits on the server
Opening Frasier on a Fairphone took ~5 s to render the episode list
although every episode was cached. Three causes:

- resolve_series_view waited for Next Up and resume before returning the
  episodes, and Next Up was server-first. The episode list now returns as
  soon as the episodes are in (with_hints); hints that have answered are
  used, late ones dropped, and the picker falls back to local watch state.
  Next Up is cache-first like every other query.
- The page loaded itself six times per open: onMount plus a mount-time
  $effect, the reachability effect's first run posing as a reconnect, and
  a double mount. All triggers now share one coalesced load per item
  (createCoalescedLoader); refresh triggers get one re-run after it.
- The root layout rendered the route in two branches that each rendered
  children; the page store deciding between them updates a flush late, so
  navigating Search -> library page mounted the page twice. One element
  now renders the route and only its classes change.

On the device: one load per open, seasons from cache in 14 ms, episodes
and the Resume button up in under a second (was ~5 s).
2026-09-24 04:45:44 +02:00
dtourolle c0545a245f perf(db): one logical container per item, indexed; no whole-table reads
Listings matched children on four columns at once (parent_id, album_id,
season_id, series_id) because Jellyfin's ParentId is the storage parent,
not the logical one. The OR defeated the planner into a full scan, and it
was wrong: every episode carries its series id, so a series listed all its
episodes beside its seasons (on both the browse and Downloads surfaces).

- Migration 027 adds items.container_id, a VIRTUAL generated column
  (episode -> season/series/parent, season -> series, track -> album,
  else parent) indexed with (sort_name, name), so a listing is one
  ordered index range and every write path is covered untouched.
- Containers never cached (an episode that arrived via Next Up) get
  placeholders named from the child's own fields, in the migration and
  on every cache write, so offline navigation stays series -> season.
- The six queries that built the set of every downloaded item in a CTE
  (get_item, latest, recently played, search, favourites, by-person,
  Downloads) now check availability per row with one shared predicate.
- PRAGMA optimize at open gives the planner statistics.

Benchmark (~110k items, desktop): series listing ~80 ms -> <1 ms;
migration 027 upgrades that database in ~0.1 s (0.2 s on the Fairphone).
Tests first: a series listing its episodes, and the Downloads series
drill, both failed before the change.
2026-09-24 04:45:44 +02:00
dtourolle e21ddae737 chore(release): v0.13.1
🏗️ Build and Test JellyTau / Run Tests (push) Skipped
🏗️ Build and Test JellyTau / Android Compile Check (push) Skipped
🏗️ Build and Test JellyTau / Supply Chain (push) Successful in 3m56s
📱 Test APK / Build test APK (push) Canceled after 9m51s
Publish Documentation / Build & publish docs to gitea-pages (push) Canceled after 0s
Traceability Validation / Check Requirement Traces (push) Successful in 13s
Build & Release / Run Tests (push) Successful in 10m30s
Build & Release / Build Linux (push) Successful in 14m30s
Build & Release / Build Windows (push) Successful in 10m56s
Build & Release / Build Android (push) Successful in 20m10s
Build & Release / Create Release (push) Successful in 33s
Pages answer from the cache again: database reads no longer wait behind
writes, the listing query uses its indexes, and a cache answer races the
server instead of waiting it out. Plus the series page's single season
walk and the background-audio return fix.
2026-09-24 03:59:10 +02:00
dtourolle 21f24dd998 perf(db): reads no longer wait behind writes; pages answer from cache
A series page took about a second to show its seasons on a phone, every
visit, although they were cached. Three things stacked up:

- One SQLite connection behind one mutex served the whole app, so every
  read queued behind every write. The database now has one owner: a
  writer thread for writes and a pool of read-only WAL connections for
  reads. synchronous = NORMAL and a busy timeout on every connection.
- The listing query built the set of every available item in the
  database before filtering to the parent (~80 ms on a desktop for a
  100k-item cache), then fetched user data one row at a time. It now
  checks availability per row, uses the hierarchy indexes (1.5 ms on
  the same benchmark) and batches the user-data lookup.
- A cache read that missed the 100 ms fast path was set aside until the
  server answered. It is now raced against the server; whichever answers
  first with content wins.

On the Fairphone, Frasier's season and episode lists now come from
cache in 34-133 ms (was 600-1030 ms waiting on the server).

Fixes found on the way, each with a test that failed first:
- sync_queue_mutation could return another mutation's row id: the id
  came from a second trip to the shared connection. insert() reads it in
  the same job.
- save_to_cache switched foreign keys off on the shared connection
  across its awaits, so concurrent writes ran unchecked. The toggle now
  lives inside one writer job, and a page is one transaction instead of
  one commit per row.

Also: thumbnail LRU touches no longer block the lookup; unused
tokio-rusqlite dropped. Design and invariants in
docs/architecture/08-database-design.md (Connection ownership, Listing
query shape) and 03-data-flow.md.
2026-09-24 03:58:04 +02:00
dtourolle 1fb5f070c8 fix(player): return from background audio onto the episode it advanced to
An episode that ends while backgrounded in audio-only mode advances in the
backend, but player_exit_background_audio returned only a position, so the
video page reloaded the episode it was mounted with -- the previous one, at
the new episode's timestamp.

The command now returns BackgroundAudioResume { itemId, positionSeconds }.
planHandoffReturn yields "other-item" when the id differs from the mounted
one, and the player page navigates to that episode with resumeAt=<seconds>,
marking the outgoing episode watched and suppressing its stale stop report.

TRACES: UR-040, UR-023 | DR-296 | UT-265, UT-266
2026-09-24 03:45:59 +02:00
dtourolle f6efa7208a perf(series): list a series' episodes with one concurrent season walk
"More info" on Frasier took ~10 s. The series page asked Rust for the
episodes and for the current episode as two commands; each walked every
season, and each walk fetched the eleven seasons one after another. So the
wait was the sum of twenty-two listings, each a cache read queued behind
whatever the database was writing — measured at ~4 s per walk on a Fairphone
5 while the launch-time catalog sync ran.

Seasons are now fetched together (gather_season_episodes), so a walk waits
for its slowest season, not the sum. And repository_get_series_view returns
the episodes and the current episode from one walk, with Next Up and resume
fetched alongside it; the series page makes that one call.

Under today's single database connection the cache reads themselves still
queue on its mutex; the concurrency pays off fully once reads get their own
connections. Halving the walks helps regardless.

Test first: ten 100 ms seasons took 1.01 s sequentially; now well under the
400 ms bound, with a failing season still leaving the rest.

DR-295, UT-264.
2026-09-24 03:22:31 +02:00
34 changed files with 6178 additions and 3710 deletions
+57
View File
@@ -9,6 +9,63 @@ generated trace matrix lives in [docs/traceability.md](docs/traceability.md).
For how long each fixed defect had been shipping before it was found, see For how long each fixed defect had been shipping before it was found, see
[docs/defect-windows.md](docs/defect-windows.md). [docs/defect-windows.md](docs/defect-windows.md).
## v0.13.2
A series opens with its episodes in under a second. v0.13.1 fixed the season
list; the episode list below it still took about five seconds on a Fairphone.
### ⚡ Performance
- **The episode list no longer waits for the server.** It used to wait for
"Next Up" and your resume point, fetched from the server first, although every
episode was already on the device. The list now shows as soon as it is
loaded, and Next Up answers from the device first like everything else.
(DR-101, DR-295)
- **A page loads once.** Opening a series loaded it six times over, putting
about seventy requests in flight at once. It now loads once, and re-loads only
when something actually changed (reconnecting, marking watched). (DR-295)
- **The local database finds things by index instead of reading everything.**
Each item now records the one container it is listed under, so a series,
season or album page is a single index lookup — under a millisecond on a
100,000-item library, where it used to scan the whole catalogue. The update
converts an existing library in well under a second, once. (DR-012, DR-013)
### 🐛 Fixes
- **A series lists its seasons, not every episode as well.** Both the library
view and the Downloads view mixed a series' episodes in with its seasons.
(DR-013)
## v0.13.1
Pages answer from the cache again. Found on a Fairphone, where opening a
series took one to ten seconds even though everything on it was already cached.
### ⚡ Performance
- **Series and library pages load from the cache.** Frasier's season and
episode lists now appear in 34–133 ms on a Fairphone 5; they took 600–1030 ms
each, waiting on the server. Three causes stacked up. Every database read
waited behind every write, because one connection served the whole app. The
listing query scanned the entire cached catalogue whatever page it was for.
And a cache answer that took longer than 100 ms was ignored until the server
replied. Reads now have their own connections, the query uses its indexes
(about 50× faster on a 100k-item cache), and the cache and the server race:
whichever answers first with something to show wins. (DR-012, DR-013)
- **A series page walks its seasons once, in parallel.** It used to walk all of
them twice, one after another. (DR-295)
### 🐛 Fixes
- **Coming back from background audio opens the right episode.** If an episode
ended while the app was in the background, returning to it reloaded the
*previous* episode at the new one's position. (DR-296)
- **Offline changes keep their own identity.** Two favourites or progress
updates queued at the same moment could be handed each other's queue entry,
so syncing one marked the other as done. (DR-014)
- **Catalog caching no longer switches off data-integrity checks for the rest
of the app** while it saves a page. (DR-012)
## v0.13.0 ## v0.13.0
Android plays and downloads the original file, and offline mode works without Android plays and downloads the original file, and offline mode works without
+35 -46
View File
@@ -314,57 +314,22 @@ pub struct PlaybackModeManager {
**Location**: `src-tauri/src/storage/db_service.rs` **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 ```rust
#[async_trait] let db_service = database.service(); // cheap clone of the shared owner
pub trait DatabaseService: Send + Sync {
async fn execute(&self, query: Query) -> Result<usize, DatabaseError>;
async fn execute_batch(&self, queries: Vec<Query>) -> Result<(), DatabaseError>;
async fn query_one<T, F>(&self, query: Query, mapper: F) -> Result<T, DatabaseError>
where F: FnOnce(&Row) -> Result<T> + Send + 'static;
async fn query_optional<T, F>(&self, query: Query, mapper: F) -> Result<Option<T>, DatabaseError>
where F: FnOnce(&Row) -> Result<T> + Send + 'static;
async fn query_many<T, F>(&self, query: Query, mapper: F) -> Result<Vec<T>, DatabaseError>
where F: Fn(&Row) -> Result<T> + Send + 'static;
async fn transaction<F, T>(&self, f: F) -> Result<T, DatabaseError>
where F: FnOnce(Transaction) -> Result<T> + Send + 'static;
}
pub struct RusqliteService {
connection: Arc<Mutex<Connection>>,
}
impl DatabaseService for RusqliteService {
async fn execute(&self, query: Query) -> Result<usize, DatabaseError> {
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 query = Query::with_params("SELECT ...", vec![...]); 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 ## Component Hierarchy
```mermaid ```mermaid
@@ -701,6 +666,14 @@ render behind it. There is no measurement and no reserved padding. If you
restructure the shell, preserve the scroll containment — reintroducing padding restructure the shell, preserve the scroll containment — reintroducing padding
math reintroduces the bug. math reintroduces the bug.
**The route renders in exactly one element.** `+layout.svelte` switches the
wrapper's *classes* between the shell scroller and the plain clipped box that
layout-owning routes (library, settings, player) get — it must not switch
between two branches that each render `children`. The page store that decides
the mode can update a flush after the new route renders, so two branches
mounted a page under one and then remounted it under the other: every
navigation between the two kinds of route loaded the page twice (DR-295).
### AccountMenu ### AccountMenu
One component for both breakpoints, anchored to the username/avatar (a real One component for both breakpoints, anchored to the username/avatar (a real
@@ -754,6 +727,22 @@ hero button labelled `Resume S2E4` / `Play S1E1`.
A season is not a destination: `/library/<seasonId>` redirects to its series A season is not a destination: `/library/<seasonId>` redirects to its series
(DR-103). Video library routes collapse to one per library (DR-105). (DR-103). Video library routes collapse to one per library (DR-105).
**The episode list never waits for Next Up or resume.** `resolve_series_view`
(`series_progress.rs`, `with_hints`) returns as soon as the episodes are in;
Next Up and resume are used if they have answered by then and dropped if not,
and `pick_current_episode` falls back to the episodes' own watch state. They
only refine which episode is current, and waiting for them held the list for
the server's 2–3 s although every episode was cached. Next Up is cache-first
like every other query (03-data-flow).
**One load per item, however many triggers.** The detail page loads through
`createCoalescedLoader` (`utils/coalescedLoader.ts`): calls for the item
already loading share that load, and callers that know the data changed
(`fresh`: reconnect, filter change, mark watched, clear history) get exactly
one re-run after it. `onMount`, a mount-time `$effect`, the reachability
effect's first run and the double mount above used to each start a full load —
six per open, about seventy requests in flight.
`episodeStrip.ts` holds the pure logic for the "More Episodes" strip, extracted `episodeStrip.ts` holds the pure logic for the "More Episodes" strip, extracted
from the component because it had three distinct bugs that markup made from the component because it had three distinct bugs that markup made
untestable: the strip collapsing to just the current episode while real siblings untestable: the strip collapsing to just the current episode while real siblings
+22 -7
View File
@@ -28,12 +28,17 @@ sequenceDiagram
Server->>Conn: mark_unreachable() (debounced) Server->>Conn: mark_unreachable() (debounced)
end 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 Cache-->>Hybrid: Result with items
Hybrid-->>Rust: Return cache result 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 Server-->>Hybrid: Fresh result
Hybrid-->>Rust: Return server 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 end
Rust-->>Client: SearchResult Rust-->>Client: SearchResult
@@ -42,11 +47,21 @@ sequenceDiagram
``` ```
**Key Points:** **Key Points:**
- Cache queries have 100ms timeout for responsiveness - Both legs start together. A cache answer with content inside 100 ms
- Server queries always run for fresh data (`CACHE_FAST_PATH`) returns at once.
- Cache wins if it has meaningful content - **The deadline does not decide the race.** A cache read still running at
- Automatic fallback to server if cache is empty/stale 100 ms is raced against the server (`HybridRepository::race_slow_cache`), and
- Background cache updates (planned) 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. - **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 ### Listing order is decided in Rust
@@ -224,10 +224,11 @@ them:
the current user — local path, direct play, item id as media source (a download 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) names no source, so the server served its default, which carries the item's id)
— and `HybridRepository::get_playback_info` consults it **first**. — and `HybridRepository::get_playback_info` consults it **first**.
- **A slow cache read is waited for, never discarded.** The cache is one SQLite - **A slow cache read is waited for, never discarded.** Reads no longer queue
connection behind one mutex, so any write in progress (the catalog sync that behind writes (see [Connection ownership](08-database-design.md#connection-ownership)),
starts at every launch, a download finishing) pushes a read past the 100 ms fast but a big query, a cold page cache or a busy reader pool can still push a read
path. `get_items`, the library list, genres and playlist items used to discard 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 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 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 with `cache_try` (which keeps it running) and `settle` on it when the server
+124
View File
@@ -243,9 +243,14 @@ CREATE TABLE items (
last_sync DATETIME, last_sync DATETIME,
UNIQUE(jellyfin_id, server_id) UNIQUE(jellyfin_id, server_id)
-- Logical container (migration 027): episode → season/series, season →
-- series, track → album, else parent. See "Listing query shape".
-- container_id TEXT GENERATED ALWAYS AS (CASE item_type … END) VIRTUAL
); );
-- Performance indexes -- Performance indexes
CREATE INDEX idx_items_container ON items(container_id, sort_name, name);
CREATE INDEX idx_items_server ON items(server_id); CREATE INDEX idx_items_server ON items(server_id);
CREATE INDEX idx_items_library ON items(library_id); CREATE INDEX idx_items_library ON items(library_id);
CREATE INDEX idx_items_parent ON items(parent_id); CREATE INDEX idx_items_parent ON items(parent_id);
@@ -594,6 +599,125 @@ flowchart LR
| Episode | ~3 KB | ~100 KB | 300 MB - 2 GB | | Episode | ~3 KB | ~100 KB | 300 MB - 2 GB |
| Full music library (5000 songs) | ~10 MB | ~250 MB | 25-75 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. `PRAGMA optimize` runs once at open, after migrations, so the planner has
statistics (see "Listing query shape").
**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. Every other cached read follows the same
rules.
- **Children are matched on `items.container_id`** (migration 027), a VIRTUAL
generated column holding the item's *logical* container: an episode's season
(else series, else parent), a season's series, a track's album, otherwise the
parent. Jellyfin's `ParentId` is the storage parent, not the logical one — in
a series without season folders an episode's `ParentId` is the series while
its `SeasonId` names a virtual season — so listings used to match on four
columns at once. That `OR` defeated the planner into walking the whole table,
and it was wrong: every episode carries its series id, so a series listed all
its episodes beside its seasons. Being generated, the column covers every
write path (cache, downloads, catalog crawl) without any of them knowing, and
cannot drift from the columns it is computed from.
- **`idx_items_container (container_id, sort_name, name)`** serves a listing as
one index range already in display order — no sort step. `sort_name` is
usually NULL in the cache (the cache never writes it), hence `name` in the
index too.
- **Containers exist even when never browsed.** A series lists its seasons, so
an episode whose season row was never cached (it arrived through Next Up or
Latest) would be unreachable from its series offline. `save_to_cache` — and
migration 027 for rows already on disk — inserts placeholders named from the
child's own fields (`season_name`, `series_name`, `album_name`) with
`synced_at` NULL, so they show only when a download makes them available; the
server's real row replaces them wholesale on the next browse.
- **Availability is a per-row `EXISTS`** (`downloaded_sql` / `available_sql`):
cached for browsing (only with the catalog-browse flag), downloaded, or a
container with a downloaded *descendant* — which is why that one check still
looks at all four link columns (a series is available through an episode two
levels down). It used to be a CTE that built the id of every available item
in the database before filtering, paying for the whole table on every call.
- **The library clause is added only for a library parent** (`is_library`),
never `OR`ed into an ordinary listing, where it forces a full scan.
- **`+i.server_id`** keeps the planner off the server index, which every row
shares. `PRAGMA optimize` at open (`analysis_limit = 400`) gives the planner
statistics, but even with them it chose that index for the old `OR`; the `+`
is the guarantee.
- **User data is fetched in batches** (`with_user_data`, one `IN (…)` query per
500 rows), not once per row.
Measured on a ~110k-item benchmark catalogue (desktop): a series listing went
from ~80 ms to under 1 ms; migration 027 upgrades an existing database of that
size in ~0.1 s. `listing_a_non_library_parent_uses_the_container_index` and
migration 027's tests assert the plans; `storage::tests::write_bench_database`
(ignored; set `JELLYTAU_BENCH_DB`) writes the benchmark catalogue for the
`sqlite3` CLI.
## Rust Module Structure ## Rust Module Structure
``` ```
+8 -3
View File
@@ -496,6 +496,8 @@ Internal architecture, components, and application logic.
| DR-292 | The offline catalog reveal is one rule, applied by both library views. Two defects, one cause — "server only" was a private `$derived` inside `MediaCard`. (1) The list view (`LibraryListView`, what `LibraryGrid` renders when the stored view preference is `list`) had no notion of it at all, so a library browsed as a list offline showed every revealed item as an ordinary tappable row that plays nothing, with no way to queue it. (2) The rule asked the downloads store whether *this item id* was downloaded, but only a playable leaf (Audio, Movie, Episode) ever has a download row — an album's tracks carry them, the album does not — so a fully downloaded album greyed itself out and offered to queue what was already on the device, which is what "my downloaded music is greyed out" was. The rule moves to the pure `$lib/utils/serverOnly`, both views call it, and the container half is answered by the backend: `get_download_disk_usage().sizes` already carries container subtotals beside leaf sizes (DR-085), so `deviceContentIds` is membership in a Rust-computed map rather than a frontend guess at which item types are containers. That map was loaded only by the Downloads page, so the shell now primes it at startup and re-reads it whenever the offline gate settles (the DR-143 signal). Queueing is shared too (`queueOfflineDownload`), since the list view had no copy to diverge from | UI | UR-052, UR-055 | Done | | DR-292 | The offline catalog reveal is one rule, applied by both library views. Two defects, one cause — "server only" was a private `$derived` inside `MediaCard`. (1) The list view (`LibraryListView`, what `LibraryGrid` renders when the stored view preference is `list`) had no notion of it at all, so a library browsed as a list offline showed every revealed item as an ordinary tappable row that plays nothing, with no way to queue it. (2) The rule asked the downloads store whether *this item id* was downloaded, but only a playable leaf (Audio, Movie, Episode) ever has a download row — an album's tracks carry them, the album does not — so a fully downloaded album greyed itself out and offered to queue what was already on the device, which is what "my downloaded music is greyed out" was. The rule moves to the pure `$lib/utils/serverOnly`, both views call it, and the container half is answered by the backend: `get_download_disk_usage().sizes` already carries container subtotals beside leaf sizes (DR-085), so `deviceContentIds` is membership in a Rust-computed map rather than a frontend guess at which item types are containers. That map was loaded only by the Downloads page, so the shell now primes it at startup and re-reads it whenever the offline gate settles (the DR-143 signal). Queueing is shared too (`queueOfflineDownload`), since the list view had no copy to diverge from | UI | UR-052, UR-055 | Done |
| DR-293 | Android plays the original file: ExoPlayer decodes AC-3, E-AC-3, DTS and TrueHD in software through the FFmpeg extension, so neither a download nor a stream needs the server to re-encode its audio. These are licensed codecs that Android does not ship — the ROD2-W09 tablet has a vendor DTS decoder and no AC-3/E-AC-3 at all — so the download policy (DR-171) judged audio against the webview's list and turned most films into a server transcode: generated as it is sent, no `Content-Length`, `Range` ignored, measured at ~1 MB/s and restarting from zero on every network blip, against a direct copy that moved a 910 MB episode in 94 s. The renderer is `DefaultRenderersFactory` in `EXTENSION_RENDERER_MODE_ON` (platform decoders first, FFmpeg for what they lack), and `CodecDetector` reports the extension's codecs beside `MediaCodecList`'s, so the device profile and the download policy — now `renderer_can_decode_audio`, DR-234's per-platform answer, instead of the webview's list — agree with what actually decodes. The webview video path is gone on Android: it decodes none of those codecs, so an original-file download would play there as a silent film; `webview_video_fallback` (Rust) is false on Android and the frontend neither offers the switch nor honours a stored "off". Linux keeps the webview fallback beside mpv native video, and with it the server transcode for undecodable audio. Rejected: re-encoding audio on the device after download — minutes of CPU and twice the disk per film, and it would not have helped streaming. The decoder is Jellyfin's `media3-ffmpeg-decoder` build (GPL-3.0; the distributed APK carries its terms, the source stays MIT) and must be versioned in step with media3 | Playback | UR-004, UR-071 | Done | | DR-293 | Android plays the original file: ExoPlayer decodes AC-3, E-AC-3, DTS and TrueHD in software through the FFmpeg extension, so neither a download nor a stream needs the server to re-encode its audio. These are licensed codecs that Android does not ship — the ROD2-W09 tablet has a vendor DTS decoder and no AC-3/E-AC-3 at all — so the download policy (DR-171) judged audio against the webview's list and turned most films into a server transcode: generated as it is sent, no `Content-Length`, `Range` ignored, measured at ~1 MB/s and restarting from zero on every network blip, against a direct copy that moved a 910 MB episode in 94 s. The renderer is `DefaultRenderersFactory` in `EXTENSION_RENDERER_MODE_ON` (platform decoders first, FFmpeg for what they lack), and `CodecDetector` reports the extension's codecs beside `MediaCodecList`'s, so the device profile and the download policy — now `renderer_can_decode_audio`, DR-234's per-platform answer, instead of the webview's list — agree with what actually decodes. The webview video path is gone on Android: it decodes none of those codecs, so an original-file download would play there as a silent film; `webview_video_fallback` (Rust) is false on Android and the frontend neither offers the switch nor honours a stored "off". Linux keeps the webview fallback beside mpv native video, and with it the server transcode for undecodable audio. Rejected: re-encoding audio on the device after download — minutes of CPU and twice the disk per film, and it would not have helped streaming. The decoder is Jellyfin's `media3-ffmpeg-decoder` build (GPL-3.0; the distributed APK carries its terms, the source stays MIT) and must be versioned in step with media3 | Playback | UR-004, UR-071 | Done |
| DR-294 | A download plays with no network. Playing a downloaded item asked the server for its `PlaybackInfo` — only to read the media-source id that subtitle URLs are keyed by — and `HybridRepository::get_playback_info` went to the server alone, so offline the call retried for seven seconds, failed, and the file on disk was never opened. A completed download for the current user now answers playback info from its download row, first and regardless of reachability: the local path, direct play, and the item id as media-source id (a download names no source, so the server served its default, which carries the item's id). Next Up had the same shape — server-only — and the TV landing page loads it in one `Promise.all` with its other rows, so offline that single failure blanked the whole page with Continue Watching and Latest sitting in the cache; it now falls back to the cache when the server cannot answer. And 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 at every launch, a download finishing) pushes a read past the 100 ms fast path, and `get_items`, the library list, genres and playlist items discarded such a read, waited on the server, and offline returned its error over data on disk — "More info" on a downloaded show failed exactly so. They keep the read running (`cache_try`) and wait for it when the server fails (`settle`); the cache-only reads (search, favourites) simply await the cache | Repository | UR-002, UR-071 | Done | | DR-294 | A download plays with no network. Playing a downloaded item asked the server for its `PlaybackInfo` — only to read the media-source id that subtitle URLs are keyed by — and `HybridRepository::get_playback_info` went to the server alone, so offline the call retried for seven seconds, failed, and the file on disk was never opened. A completed download for the current user now answers playback info from its download row, first and regardless of reachability: the local path, direct play, and the item id as media-source id (a download names no source, so the server served its default, which carries the item's id). Next Up had the same shape — server-only — and the TV landing page loads it in one `Promise.all` with its other rows, so offline that single failure blanked the whole page with Continue Watching and Latest sitting in the cache; it now falls back to the cache when the server cannot answer. And 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 at every launch, a download finishing) pushes a read past the 100 ms fast path, and `get_items`, the library list, genres and playlist items discarded such a read, waited on the server, and offline returned its error over data on disk — "More info" on a downloaded show failed exactly so. They keep the read running (`cache_try`) and wait for it when the server fails (`settle`); the cache-only reads (search, favourites) simply await the cache | Repository | UR-002, UR-071 | Done |
| DR-295 | A series page lists its episodes with one concurrent season fan-out. "More info" on Frasier took ~10 s: the page asked Rust for the episodes and for the current episode as two commands, each of which walked every season, and each walk fetched the eleven seasons one after another — so the wait was the sum of twenty-two listings, each a cache read slowed by whatever the database was writing (the catalog sync at launch measured it at ~4 s per walk). The seasons are now fetched together (`gather_season_episodes`, so the wait is the slowest season), and `repository_get_series_view` returns the episodes and the current episode from one walk, with Next Up and resume fetched alongside it | Repository | UR-062 | Done |
| DR-296 | Returning from background audio resumes the item the native player is actually on, not the one the video page was mounted with. An episode that ends while backgrounded advances in the backend (`advance_to_next_episode_audio_only`), but `player_exit_background_audio` returned only a position, so the webview reloaded the *previous* episode at the new episode's timestamp. The command now returns `BackgroundAudioResume { itemId, positionSeconds }` (`PlayerController::background_audio_resume`); `planHandoffReturn` yields `other-item` when the id differs from the mounted one, and the player page navigates to that episode with `resumeAt=<seconds>`, recording the outgoing episode as watched and suppressing the stale unmount stop report | Playback | UR-040, UR-023 | Done (pending device verification) |
--- ---
@@ -527,7 +529,7 @@ Internal architecture, components, and application logic.
| UR-020 | IR-016, IR-018 | DR-023, DR-176 | <!-- IR-018 delivered by ExoPlayer + HTML5 `<track>`, not libmpv --> | UR-020 | IR-016, IR-018 | DR-023, DR-176 | <!-- IR-018 delivered by ExoPlayer + HTML5 `<track>`, not libmpv -->
| UR-021 | IR-016, IR-019 | DR-024 | <!-- IR-019 delivered by ExoPlayer + HLS stream re-open, not libmpv --> | UR-021 | IR-016, IR-019 | DR-024 | <!-- IR-019 delivered by ExoPlayer + HLS stream re-open, not libmpv -->
| UR-022 | IR-017 | DR-025 | | UR-022 | IR-017 | DR-025 |
| UR-023 | IR-010 | DR-026, DR-047, DR-048, DR-049, DR-263 | | UR-023 | IR-010 | DR-026, DR-047, DR-048, DR-049, DR-263, DR-296 |
| UR-024 | IR-010 | DR-027 | | UR-024 | IR-010 | DR-027 |
| UR-025 | IR-015 | DR-028, DR-131, DR-132, DR-178, DR-179 | | UR-025 | IR-015 | DR-028, DR-131, DR-132, DR-178, DR-179 |
| UR-026 | - | DR-029, DR-048, DR-050 | | UR-026 | - | DR-029, DR-048, DR-050 |
@@ -544,7 +546,7 @@ Internal architecture, components, and application logic.
| UR-037 | IR-010 | DR-042 | | UR-037 | IR-010 | DR-042 |
| UR-038 | IR-010 | DR-043 | | UR-038 | IR-010 | DR-043 |
| UR-039 | - | DR-045, DR-046 | | UR-039 | - | DR-045, DR-046 |
| UR-040 | IR-025 | DR-051, DR-052, DR-129, DR-130, DR-159, DR-178, DR-179, DR-180, DR-183, DR-190, DR-196, DR-201, DR-203, DR-263, DR-266 | | UR-040 | IR-025 | DR-051, DR-052, DR-129, DR-130, DR-159, DR-178, DR-179, DR-180, DR-183, DR-190, DR-196, DR-201, DR-203, DR-263, DR-266, DR-296 |
| UR-041 | IR-026 | DR-053, DR-160, DR-161, DR-172, DR-182, DR-183, DR-184, DR-185, DR-188, DR-265, DR-266 | | UR-041 | IR-026 | DR-053, DR-160, DR-161, DR-172, DR-182, DR-183, DR-184, DR-185, DR-188, DR-265, DR-266 |
| UR-042 | IR-009, IR-014 | DR-054 | | UR-042 | IR-009, IR-014 | DR-054 |
| UR-043 | IR-027 | DR-055, DR-291 | | UR-043 | IR-027 | DR-055, DR-291 |
@@ -565,7 +567,7 @@ Internal architecture, components, and application logic.
| UR-058 | - | DR-087, DR-142 | | UR-058 | - | DR-087, DR-142 |
| UR-060 | - | DR-090, DR-091, DR-111 | | UR-060 | - | DR-090, DR-091, DR-111 |
| UR-061 | - | DR-092 | | UR-061 | - | DR-092 |
| UR-062 | - | DR-101, DR-102, DR-103, DR-104, DR-107 | | UR-062 | - | DR-101, DR-102, DR-103, DR-104, DR-107, DR-295 |
| UR-063 | - | DR-105 | | UR-063 | - | DR-105 |
| UR-064 | - | DR-106 | | UR-064 | - | DR-106 |
| UR-065 | IR-030 | DR-108, DR-109, DR-110, DR-111 | | UR-065 | IR-030 | DR-108, DR-109, DR-110, DR-111 |
@@ -854,6 +856,9 @@ Internal architecture, components, and application logic.
| UT-261 | Next Up answers from the cache, rather than failing, when the server is unreachable | DR-294 | Done | | UT-261 | Next Up answers from the cache, rather than failing, when the server is unreachable | DR-294 | Done |
| UT-262 | The Android webview fallback is neither offered in Settings nor honoured by the player unless Rust reports it, so a stored "native video off" cannot route video to a renderer that plays the original file silent | DR-293 | Done | | UT-262 | The Android webview fallback is neither offered in Settings nor honoured by the player unless Rust reports it, so a stored "native video off" cannot route video to a renderer that plays the original file silent | DR-293 | Done |
| UT-263 | With the database held past the 100 ms fast path and the server unreachable, `get_items`, the library list, a cache-only search and cache-only favourites all answer from the cache instead of failing | DR-294 | Done | | UT-263 | With the database held past the 100 ms fast path and the server unreachable, `get_items`, the library list, a cache-only search and cache-only favourites all answer from the cache instead of failing | DR-294 | Done |
| UT-264 | Ten seasons whose listings each take 100 ms are gathered in well under the 1 s a sequential walk takes, and a season that fails to load leaves the other nine seasons' episodes in the result | DR-295 | Done |
| UT-265 | `planHandoffReturn` switches to the item the backend advanced to while backgrounded, and reloads in place when the backend is still on the mounted item or reports none | DR-296 | Done |
| UT-266 | After a background-audio episode advance, the controller's resume point names the new episode and carries no base from the previous one | DR-296 | Done |
### Integration Tests ### Integration Tests
| Test ID | Test Description | Traces To | Status | | Test ID | Test Description | Traces To | Status |
+3042 -2627
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "jellytau", "name": "jellytau",
"version": "0.13.0", "version": "0.13.2",
"description": "A cross-platform Jellyfin client built with Tauri, SvelteKit and Rust.", "description": "A cross-platform Jellyfin client built with Tauri, SvelteKit and Rust.",
"author": "Duncan Tourolle <duncan@tourolle.paris>", "author": "Duncan Tourolle <duncan@tourolle.paris>",
"license": "MIT", "license": "MIT",
+1 -13
View File
@@ -2275,7 +2275,7 @@ dependencies = [
[[package]] [[package]]
name = "jellytau" name = "jellytau"
version = "0.13.0" version = "0.13.2"
dependencies = [ dependencies = [
"aes-gcm", "aes-gcm",
"argon2", "argon2",
@@ -2315,7 +2315,6 @@ dependencies = [
"tempfile", "tempfile",
"tiny_http", "tiny_http",
"tokio", "tokio",
"tokio-rusqlite",
"tokio-util", "tokio-util",
"urlencoding", "urlencoding",
"uuid", "uuid",
@@ -5340,17 +5339,6 @@ dependencies = [
"syn 2.0.112", "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]] [[package]]
name = "tokio-rustls" name = "tokio-rustls"
version = "0.26.4" version = "0.26.4"
+1 -2
View File
@@ -4,7 +4,7 @@ name = "jellytau"
# `player-conformance`, and a second binary makes a bare `cargo run` — # `player-conformance`, and a second binary makes a bare `cargo run` —
# which `tauri dev` issues — ambiguous. # which `tauri dev` issues — ambiguous.
default-run = "jellytau" default-run = "jellytau"
version = "0.13.0" version = "0.13.2"
description = "A cross-platform Jellyfin client" description = "A cross-platform Jellyfin client"
authors = ["Duncan Tourolle <duncan@tourolle.paris>"] authors = ["Duncan Tourolle <duncan@tourolle.paris>"]
license = "MIT" license = "MIT"
@@ -53,7 +53,6 @@ futures-util = "0.3"
async-trait = "0.1" async-trait = "0.1"
# SQLite for offline storage # SQLite for offline storage
tokio-rusqlite = "0.6"
rusqlite = { version = "0.32", features = ["bundled"] } rusqlite = { version = "0.32", features = ["bundled"] }
chrono = { version = "0.4", features = ["serde"] } chrono = { version = "0.4", features = ["serde"] }
directories = "5" directories = "5"
+12 -7
View File
@@ -939,12 +939,17 @@ pub async fn player_background_action(
Ok(action) Ok(action)
} }
/// TRACES: UR-040 | DR-052 | UT-061, IT-013 /// Returns the item the native player is on and its absolute position. The
/// item matters: an episode that ended while backgrounded has already advanced
/// in the backend, so reloading the video the webview was mounted with would
/// bring back the previous episode. (DR-296)
///
/// TRACES: UR-040, UR-023 | DR-052, DR-296 | UT-061, IT-013
#[tauri::command] #[tauri::command]
#[specta::specta] #[specta::specta]
pub async fn player_exit_background_audio( pub async fn player_exit_background_audio(
player: State<'_, PlayerStateWrapper>, player: State<'_, PlayerStateWrapper>,
) -> Result<f64, String> { ) -> Result<crate::player::BackgroundAudioResume, String> {
let controller = player.0.lock().await; let controller = player.0.lock().await;
// Read the position BEFORE clearing either base. The position tick applies the // Read the position BEFORE clearing either base. The position tick applies the
@@ -954,23 +959,23 @@ pub async fn player_exit_background_audio(
// lock discipline from CLAUDE.md: never hold work across a re-entrant call. // lock discipline from CLAUDE.md: never hold work across a re-entrant call.
// (DR-159) // (DR-159)
// //
// `absolute_position` rather than `position`, because a tick that has not // `background_audio_resume` reads `absolute_position` rather than `position`, because a tick that has not
// landed *yet* is the same hazard from the other side: returning to the // landed *yet* is the same hazard from the other side: returning to the
// foreground while the audio-only transcode is still opening read 0.0, and // foreground while the audio-only transcode is still opening read 0.0, and
// the video reloaded at StartTimeTicks=0 — the episode restarting from the // the video reloaded at StartTimeTicks=0 — the episode restarting from the
// beginning. Flooring at the handoff base cannot overshoot: the stream is // beginning. Flooring at the handoff base cannot overshoot: the stream is
// physically incapable of being behind its own starting point. (DR-178) // physically incapable of being behind its own starting point. (DR-178)
let absolute = controller.absolute_position(); let resume = controller.background_audio_resume();
// Now safe to tear the handoff down, native side first. // Now safe to tear the handoff down, native side first.
let _ = crate::player::set_lockscreen_position_offset(0.0); let _ = crate::player::set_lockscreen_position_offset(0.0);
controller.exit_background_audio(); controller.exit_background_audio();
controller.stop().map_err(|e| e.to_string())?; controller.stop().map_err(|e| e.to_string())?;
info!( info!(
"player_exit_background_audio: resuming the video at {:.1}s", "player_exit_background_audio: resuming {:?} at {:.1}s",
absolute resume.item_id, resume.position_seconds
); );
Ok(absolute) Ok(resume)
} }
/// Play a queue of media items /// Play a queue of media items
+18
View File
@@ -514,6 +514,24 @@ pub async fn repository_get_series_current_episode(
.map_err(|e| format!("{:?}", e)) .map_err(|e| format!("{:?}", e))
} }
/// A series' episodes and the viewer's current episode, from one season
/// fan-out. The series page used to ask for these as two commands, each of
/// which walked every season.
///
/// TRACES: UR-062 | DR-101, DR-295
#[tauri::command]
#[specta::specta]
pub async fn repository_get_series_view(
manager: State<'_, RepositoryManagerWrapper>,
handle: String,
series_id: String,
) -> Result<series_progress::SeriesView, String> {
let repo = manager.0.get(&handle).ok_or("Repository not found")?;
series_progress::resolve_series_view(repo.as_ref(), &series_id)
.await
.map_err(|e| format!("{:?}", e))
}
/// Erase the viewer's watch history for an item. /// Erase the viewer's watch history for an item.
/// ///
/// Clears the played flag and the resume position; on a series or season the /// Clears the played flag and the resume position; on a series or season the
+70 -7
View File
@@ -49,6 +49,19 @@ pub async fn sync_queue_mutation(
Arc::new(database.service()) 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<S: DatabaseService>(
db_service: &S,
user_id: String,
operation: String,
item_id: Option<String>,
payload: Option<String>,
) -> Result<i64, String> {
let query = Query::with_params( let query = Query::with_params(
"INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at) "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at)
VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP)", 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())?; // `insert`, not `execute` + `last_insert_rowid`: the id must be read in
let id = db_service // the same job as the insert, or a concurrent write hands us its row.
.last_insert_rowid() db_service.insert(query).await
.await
.map_err(|e| e.to_string())?;
Ok(id)
} }
/// Get all pending sync operations for a user /// Get all pending sync operations for a user
@@ -387,4 +396,58 @@ mod tests {
assert!(item.retry_count == i); 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}");
}
}
} }
+2
View File
@@ -238,6 +238,7 @@ use commands::{
repository_get_resume_movies, repository_get_resume_movies,
repository_get_series_current_episode, repository_get_series_current_episode,
repository_get_series_episodes, repository_get_series_episodes,
repository_get_series_view,
repository_get_similar_items, repository_get_similar_items,
repository_get_stream_selection, repository_get_stream_selection,
repository_get_subtitle_url, repository_get_subtitle_url,
@@ -1014,6 +1015,7 @@ fn specta_builder() -> Builder<tauri::Wry> {
repository_get_next_up_episodes, repository_get_next_up_episodes,
repository_get_series_episodes, repository_get_series_episodes,
repository_get_series_current_episode, repository_get_series_current_episode,
repository_get_series_view,
repository_clear_watch_history, repository_clear_watch_history,
repository_get_recently_played_audio, repository_get_recently_played_audio,
repository_get_resume_movies, repository_get_resume_movies,
+77
View File
@@ -177,6 +177,20 @@ fn completion_report_position(runtime: Option<f64>, last_position: f64) -> f64 {
#[cfg_attr(not(target_os = "android"), allow(dead_code))] #[cfg_attr(not(target_os = "android"), allow(dead_code))]
const RESUME_BACKOFF_STEP_SECS: u64 = 2; const RESUME_BACKOFF_STEP_SECS: u64 = 2;
/// Where playback stands when a background-audio handoff returns to the
/// foreground. See [`PlayerController::background_audio_resume`].
///
/// TRACES: UR-040, UR-023 | DR-296
#[derive(specta::Type, Debug, Clone, PartialEq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct BackgroundAudioResume {
/// Item the native audio player is on — `None` if the queue emptied (e.g.
/// the sleep timer stopped playback while backgrounded).
pub item_id: Option<String>,
/// Absolute position in that item, in seconds.
pub position_seconds: f64,
}
/// Metadata for the lockscreen / media notification. /// Metadata for the lockscreen / media notification.
/// ///
/// Used to drive the Android MediaSession from Rust in remote (cast) mode, where /// Used to drive the Android MediaSession from Rust in remote (cast) mode, where
@@ -1728,6 +1742,23 @@ impl PlayerController {
/// reports are honoured from here on. /// reports are honoured from here on.
/// ///
/// TRACES: UR-040, UR-005 | DR-052, DR-097 /// TRACES: UR-040, UR-005 | DR-052, DR-097
/// Where the foreground should pick up from a background-audio handoff: the
/// item the native player is on now, and its absolute position.
///
/// The item is not necessarily the one the handoff started from — an episode
/// that ends while backgrounded advances in the backend
/// (`advance_to_next_episode_audio_only`) — so the webview must not assume
/// it can reload the video it was mounted with. Read-only: call it before
/// `exit_background_audio` clears the base the position depends on.
///
/// TRACES: UR-040, UR-023 | DR-296 | UT-266
pub fn background_audio_resume(&self) -> BackgroundAudioResume {
BackgroundAudioResume {
item_id: self.queue.lock_safe().current().map(|item| item.id.clone()),
position_seconds: self.absolute_position(),
}
}
pub fn exit_background_audio(&self) -> f64 { pub fn exit_background_audio(&self) -> f64 {
*self.background_audio_active.lock_safe() = false; *self.background_audio_active.lock_safe() = false;
self.take_background_audio_base() self.take_background_audio_base()
@@ -4411,6 +4442,52 @@ mod tests {
); );
} }
/// Returning to the foreground after the backend advanced to the next episode
/// must bring back THAT episode, not the one the handoff started from.
///
/// The return used to carry only a position; the webview reloaded the video it
/// was mounted with, so the user came back to the previous episode — at the new
/// episode's timestamp. The resume point therefore names the item the native
/// player is actually on.
///
/// TRACES: UR-040 | DR-296 | UT-266
#[tokio::test]
async fn test_background_audio_resume_names_the_advanced_episode() {
let controller = PlayerController::default();
controller.set_repository(Arc::new(MockEpisodeRepo::season(3)));
let episode = MediaItem {
transport: None,
id: "ep1".to_string(),
item_type: Some("Episode".to_string()),
media_type: MediaType::Audio,
series_id: Some("series1".to_string()),
..create_test_items(1).remove(0)
};
controller.play_queue(vec![episode], 0).unwrap();
controller.enter_background_audio(1200.0);
let before = controller.background_audio_resume();
assert_eq!(before.item_id.as_deref(), Some("ep1"));
assert_eq!(before.position_seconds, 1200.0);
controller
.advance_to_next_episode_audio_only("ep2")
.await
.expect("advance should succeed");
let after = controller.background_audio_resume();
assert_eq!(
after.item_id.as_deref(),
Some("ep2"),
"the foreground must resume the episode the backend advanced to"
);
assert!(
after.position_seconds < 1200.0,
"the previous episode's handoff base must not leak into the new one"
);
}
/// A background audio-only episode must advance IN THE BACKEND when the /// A background audio-only episode must advance IN THE BACKEND when the
/// autoplay decision comes back as ShowNextEpisodePopup — never by starting a /// autoplay decision comes back as ShowNextEpisodePopup — never by starting a
/// countdown the frontend is supposed to act on. /// countdown the frontend is supposed to act on.
+338 -102
View File
@@ -50,6 +50,42 @@ impl<T> CacheLeg<T> {
} }
} }
/// 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<SearchResult, RepoError>,
) {
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<T> {
/// 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 /// Hybrid repository combining online and offline data sources
/// ///
/// Uses cache-first parallel racing strategy: /// 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<T, F>(
mut slow: tokio::task::JoinHandle<Result<T, RepoError>>,
server: F,
) -> Raced<T>
where
T: MeaningfulContent + ExcludeHidden,
F: std::future::Future<Output = Result<T, RepoError>>,
{
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. /// 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) /// 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 { match server_future.await {
Ok(data) => Ok(data.without_excluded()), Ok(data) => Ok(data.without_excluded()),
Err(e) => { // Cache answered in time but had nothing: return that, so an
// The server cannot answer. If the cache is still working, it is // empty-but-valid cached listing still beats a network error.
// now the only thing that can, so wait it out rather than Err(e) => fast.unwrap_or(Err(e)),
// 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))
}
} }
} }
@@ -498,10 +587,10 @@ impl HybridRepository {
/// ///
/// Missing the deadline does **not** cancel the query — it keeps running on /// Missing the deadline does **not** cancel the query — it keeps running on
/// its own task and [`CacheLeg::settle`] can still collect it. That /// its own task and [`CacheLeg::settle`] can still collect it. That
/// distinction is the whole point. The database is one SQLite connection /// distinction is the whole point. A read can miss the deadline for many
/// behind one mutex, so a concurrent write (a sync drain, a bulk /// reasons — a large listing, slow storage, a busy reader pool (and, before
/// `save_to_cache`) blocks reads for its duration and this deadline trips /// reads had their own connections, any write in progress). Treating that
/// routinely on slow storage. Treating that as "the cache is empty" while /// as "the cache is empty" while
/// throwing the answer away meant that offline — where the server leg also /// throwing the answer away meant that offline — where the server leg also
/// fails — browsing surfaced a network error instead of the cached content /// fails — browsing surfaced a network error instead of the cached content
/// sitting right there on disk. /// sitting right there on disk.
@@ -532,9 +621,9 @@ impl HybridRepository {
/// if it made it, and otherwise the read itself, still running. /// if it made it, and otherwise the read itself, still running.
/// ///
/// The cache-then-server queries used to *discard* a read that missed the /// 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 /// deadline. When reads shared one connection with writes, any write in
/// write in progress — the catalog sync that starts at every launch, a /// progress — the catalog sync that starts at every launch, a download
/// download finishing — pushes a read past 100 ms routinely; offline the /// finishing — pushed a read past 100 ms routinely; offline the
/// server then failed too, and the page reported a network error over data /// 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. /// sitting on disk. Keeping the read lets [`Self::settle`] wait for it.
/// ///
@@ -576,6 +665,48 @@ impl HybridRepository {
fast.or(Err(server_err)) 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<Result<SearchResult, RepoError>>,
offline: Arc<OfflineRepository>,
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<OfflineRepository>,
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 /// [`Self::settle`] for `get_items`, whose slow read has not yet had
/// exclusions applied. /// exclusions applied.
async fn cache_or( async fn cache_or(
@@ -653,8 +784,12 @@ impl MediaRepository for HybridRepository {
let opts_clone = options.clone(); let opts_clone = options.clone();
// Start server request in background (non-blocking) // Start server request in background (non-blocking)
let server_handle = let mut server_handle = tokio::spawn(async move {
tokio::spawn(async move { online.get_items(&parent_id_clone, options).await }); 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). // 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") // is what made a downloaded show fail offline ("Failed to load item")
// whenever a write held the database past 100 ms — the catalog sync // whenever a write held the database past 100 ms — the catalog sync
// that starts at every launch does, routinely. TRACES: UR-002 | DR-294 // that starts at every launch does, routinely. TRACES: UR-002 | DR-294
let (cache_result, slow_cache) = let (cache_result, slow_cache) = Self::cache_try(async move {
Self::cache_try(async move { offline.get_items(&parent_id, opts_clone).await }).await; 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); let cache_result = cache_result.map(ExcludeHidden::without_excluded);
// Downloads-only gate: when the "Show all server media" toggle is off // 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 {}", "[HybridRepo] Cache hit for get_items, returning immediately for parent {}",
&parent_id_for_save[..8.min(parent_id_for_save.len())] &parent_id_for_save[..8.min(parent_id_for_save.len())]
); );
// Background: save server result to cache when it arrives Self::save_when_server_answers(server_handle, offline_for_save, parent_id_for_save);
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
}
});
return Ok(data.clone()); 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 { match server_handle.await {
Ok(Ok(server_data)) => { Ok(Ok(server_data)) => {
if !server_data.items.is_empty() { Self::save_in_background(offline_for_save, parent_id_for_save, &server_data);
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())]
);
}
});
}
// The cache keeps the server's full page (above) — an exclusion // The cache keeps the server's full page (above) — an exclusion
// is a view preference and can be undone, so hiding items from // is a view preference and can be undone, so hiding items from
// the *cache* would make un-hiding them require a re-crawl. Only // 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 // TRACES: UR-076 | DR-209
Ok(server_data.without_excluded()) 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) => { Err(join_err) => {
let e = RepoError::Network { let e = RepoError::Network {
message: format!("Server task failed: {}", join_err), 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. /// background so the stored copy keeps up with the server.
/// ///
/// The background refresh is what carries per-user state home: caching an /// 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 /// item also writes its `user_data_mirror_query` row, the only path by
/// position set on another device reaches the local `user_data` row the /// which a watch position set on another device reaches the local
/// resume check reads. Without it a cache hit returned this device's own /// `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 — /// stale position forever and cross-device resume silently did nothing —
/// `get_items` already refreshes this way, so browsing a season worked /// `get_items` already refreshes this way, so browsing a season worked
/// while opening the episode directly did not. /// while opening the episode directly did not.
@@ -869,22 +1000,32 @@ impl MediaRepository for HybridRepository {
series_id: Option<&str>, series_id: Option<&str>,
limit: Option<usize>, limit: Option<usize>,
) -> Result<Vec<MediaItem>, RepoError> { ) -> Result<Vec<MediaItem>, RepoError> {
// Next Up is dynamic, so the server's answer is preferred — but when the // Cache-first like every other query: the local answer is computed
// server cannot answer, the cache's stands in. It used to be server-only, // from the same watch state the cache refreshes from the server in the
// and the TV landing page loads Next Up in one `Promise.all` with its // background (`user_data_mirror_query`), and whichever answers first
// other rows, so offline that single failure blanked the whole page with // with content wins. It used to wait for the server outright, which
// Continue Watching and Latest sitting in the cache (DR-294). // held the series page's episode list for 2-3 s on a phone. An empty
// TRACES: UR-002 | DR-294 | UT-261 // local answer still defers to the server, and a failed server falls
match self.online.get_next_up_episodes(series_id, limit).await { // back to the cache — offline, the TV page's Next Up row must not blank
Ok(items) => Ok(items.without_excluded()), // the page (DR-294).
Err(e) => { // TRACES: UR-002 | DR-013, DR-294 | UT-261
debug!("[HybridRepo] Next Up from server failed ({e}); using the cache"); let offline = Arc::clone(&self.offline);
self.offline let online = Arc::clone(&self.online);
.get_next_up_episodes(series_id, limit) let series = series_id.map(str::to_string);
.await let series_for_server = series.clone();
.map(ExcludeHidden::without_excluded)
} let cache_future =
} Self::cache_leg(
async move { offline.get_next_up_episodes(series.as_deref(), limit).await },
)
.await;
let server_future = async move {
online
.get_next_up_episodes(series_for_server.as_deref(), limit)
.await
};
Self::parallel_race(cache_future, server_future).await
} }
async fn get_recently_played_audio( async fn get_recently_played_audio(
@@ -1385,9 +1526,9 @@ mod tests {
/// Offline, a cache read slowed past the fast path must still answer. /// Offline, a cache read slowed past the fast path must still answer.
/// ///
/// The database is one SQLite connection behind one mutex, so a concurrent /// The 100 ms fast path trips routinely on slow storage (and, while reads
/// write blocks reads for its duration and the 100 ms fast path trips on /// shared one connection with writes, behind any write). The deadline used
/// slow storage. The deadline used to *cancel* the read and report it as a /// to *cancel* the read and report it as a
/// miss; with the server leg also failing (offline), the user got a network /// miss; with the server leg also failing (offline), the user got a network
/// error over cached content that was sitting on disk. /// error over cached content that was sitting on disk.
/// ///
@@ -1464,6 +1605,101 @@ mod tests {
assert!(matches!(err, RepoError::Network { .. }), "got {err:?}"); 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::<MediaItem>::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 /// Mock offline repository that tracks queries and saves
struct MockOfflineRepo { struct MockOfflineRepo {
items: Arc<Mutex<Vec<MediaItem>>>, items: Arc<Mutex<Vec<MediaItem>>>,
File diff suppressed because it is too large Load Diff
+207 -15
View File
@@ -209,21 +209,16 @@ pub async fn fetch_series_episodes(
) -> Result<Vec<MediaItem>, RepoError> { ) -> Result<Vec<MediaItem>, RepoError> {
let children = repo.get_items(series_id, list_options()).await?; let children = repo.get_items(series_id, list_options()).await?;
let mut episodes: Vec<MediaItem> = Vec::new(); let seasons: Vec<MediaItem> = children
for season in children.items.iter().filter(|i| is_season(i)) { .items
// One failing season must not blank the whole show. .iter()
match repo.get_items(&season.id, list_options()).await { .filter(|i| is_season(i))
Ok(result) => episodes.extend(result.items.into_iter().filter(is_episode)), .cloned()
Err(e) => { .collect();
log::warn!( let mut episodes = gather_season_episodes(&seasons, |season_id| async move {
"[series] season {} of {} failed to load: {:?}", repo.get_items(&season_id, list_options()).await
season.id, })
series_id, .await;
e
);
}
}
}
// Flat series: the children *are* the episodes. // Flat series: the children *are* the episodes.
if episodes.is_empty() { if episodes.is_empty() {
@@ -234,6 +229,114 @@ pub async fn fetch_series_episodes(
Ok(episodes) Ok(episodes)
} }
/// Every episode in `seasons`, fetched with `fetch_season` (a season id → its
/// children), **all at once**.
///
/// They used to be fetched one after another, so the wait was the sum of every
/// season's listing: ~4 s for Frasier's eleven, on a phone whose cache reads
/// were slowed by a catalog sync writing in the background. Concurrently it is
/// the slowest single season. Order is restored afterwards by
/// `sort_series_order`, so completion order does not matter.
///
/// One failing season must not blank the whole show: its episodes are left out
/// and the rest returned.
///
/// TRACES: UR-062 | DR-295 | UT-264
pub async fn gather_season_episodes<F, Fut>(
seasons: &[MediaItem],
fetch_season: F,
) -> Vec<MediaItem>
where
F: Fn(String) -> Fut,
Fut: std::future::Future<Output = Result<super::SearchResult, RepoError>>,
{
let results = futures_util::future::join_all(
seasons.iter().map(|season| fetch_season(season.id.clone())),
)
.await;
let mut episodes = Vec::new();
for (season, result) in seasons.iter().zip(results) {
match result {
Ok(result) => episodes.extend(result.items.into_iter().filter(is_episode)),
Err(e) => log::warn!("[series] season {} failed to load: {:?}", season.id, e),
}
}
episodes
}
/// A series' episodes and the one the viewer is up to, from **one** season
/// fan-out.
///
/// The series page needs both, and asked for them as two commands; each walked
/// every season, so every visit listed the show twice. One call, one walk.
///
/// TRACES: UR-062 | DR-101, DR-295
#[derive(Debug, Clone, serde::Serialize, specta::Type)]
#[serde(rename_all = "camelCase")]
pub struct SeriesView {
pub episodes: Vec<MediaItem>,
pub current: Option<MediaItem>,
}
/// Build a [`SeriesView`]: one fan-out, with Next Up and resume fetched
/// alongside it rather than after.
///
/// TRACES: UR-062 | DR-101, DR-295
pub async fn resolve_series_view(
repo: &dyn MediaRepository,
series_id: &str,
) -> Result<SeriesView, RepoError> {
let (episodes, (next_up, resume)) = with_hints(fetch_series_episodes(repo, series_id), async {
futures_util::join!(
async {
repo.get_next_up_episodes(Some(series_id), Some(1))
.await
.unwrap_or_default()
},
async {
repo.get_resume_items(Some(series_id), Some(10))
.await
.unwrap_or_default()
},
)
})
.await;
let episodes = episodes?;
let current = pick_current_episode(series_id, &episodes, &next_up, &resume);
Ok(SeriesView { episodes, current })
}
/// Run `primary` and `hints` together, but never hold `primary` back for
/// `hints`: once `primary` is ready, the hints are taken if they have already
/// answered and dropped (`H::default()`) if not.
///
/// For the series view the primary is the episode list and the hints are Next
/// Up and resume, which only refine which episode is "current" — and the
/// picker falls back to the episodes' own watch state without them. Waiting
/// for them made the episode list wait for the server (2-3 s on a phone)
/// although every episode was in the cache in 50 ms. The cache legs of the
/// hints usually answer before the episodes do, so they are normally kept.
///
/// TRACES: UR-062 | DR-101, DR-295
async fn with_hints<P, H>(
primary: impl std::future::Future<Output = P>,
hints: impl std::future::Future<Output = H>,
) -> (P, H)
where
H: Default,
{
use futures_util::future::{select, Either};
use futures_util::FutureExt;
let primary = std::pin::pin!(primary);
let hints = std::pin::pin!(hints);
match select(primary, hints).await {
Either::Left((primary, hints)) => (primary, hints.now_or_never().unwrap_or_default()),
Either::Right((hints, primary)) => (primary.await, hints),
}
}
/// Resolve the current episode, fetching everything the policy needs. /// Resolve the current episode, fetching everything the policy needs.
/// ///
/// Next Up and resume are best-effort: offline they fail or come back empty, and /// Next Up and resume are best-effort: offline they fail or come back empty, and
@@ -293,6 +396,95 @@ mod tests {
} }
} }
/// "More info" on Frasier took ~4 s to list its episodes, twice over: the
/// eleven seasons were fetched one after another, so the wait was the *sum*
/// of eleven listings. Fetched together it is the slowest one.
///
/// TRACES: UR-062 | DR-295 | UT-264
#[tokio::test]
async fn seasons_are_fetched_concurrently_not_one_after_another() {
let seasons: Vec<MediaItem> = (1..=10)
.map(|n| MediaItem {
id: format!("season-{n}"),
item_type: "Season".to_string(),
..Default::default()
})
.collect();
let started = std::time::Instant::now();
let episodes = gather_season_episodes(&seasons, |season_id| async move {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
if season_id == "season-3" {
// One failing season must not blank the show.
return Err(RepoError::Network {
message: "gone".to_string(),
});
}
let n: i32 = season_id.trim_start_matches("season-").parse().unwrap();
Ok(crate::repository::SearchResult {
items: vec![episode(&format!("e{n}"), n, 1)],
total_record_count: 1,
})
})
.await;
let elapsed = started.elapsed();
assert!(
elapsed < std::time::Duration::from_millis(400),
"ten 100 ms seasons took {elapsed:?} — fetched in sequence, not together"
);
assert_eq!(episodes.len(), 9, "every season but the failing one");
}
/// The episode list must not wait for Next Up or resume.
///
/// The series page rendered its episodes only once Next Up had come back
/// from the server — 2-3 s on a phone while the page's other requests were
/// in flight — although every episode was in the cache after 50 ms. Those
/// two only refine which episode is "current", and the picker falls back
/// to the episodes' own watch state without them.
///
/// TRACES: UR-062 | DR-101, DR-295
#[tokio::test]
async fn the_episode_list_does_not_wait_for_slow_hints() {
let started = std::time::Instant::now();
let (episodes, hints) = with_hints(
async {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
vec![episode("e1", 1, 1)]
},
async {
tokio::time::sleep(std::time::Duration::from_millis(2000)).await;
vec![episode("from-server", 1, 2)]
},
)
.await;
let elapsed = started.elapsed();
assert_eq!(episodes.len(), 1);
assert!(hints.is_empty(), "late hints are dropped, not waited for");
assert!(
elapsed < std::time::Duration::from_millis(500),
"the episode list waited {elapsed:?} for Next Up / resume"
);
}
/// Hints that are already in (a cache answer) are used.
///
/// TRACES: UR-062 | DR-101, DR-295
#[tokio::test]
async fn hints_that_answer_first_are_kept() {
let (_, hints) = with_hints(
async {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
vec![episode("e1", 1, 1)]
},
async { vec![episode("cached", 1, 2)] },
)
.await;
assert_eq!(hints.len(), 1);
}
fn watched(mut item: MediaItem) -> MediaItem { fn watched(mut item: MediaItem) -> MediaItem {
item.user_data = Some(UserData { item.user_data = Some(UserData {
is_played: Some(true), is_played: Some(true),
+268 -106
View File
@@ -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 //! Every query in the app goes through [`RusqliteService`], which owns the
//! the underlying database implementation. This makes it easy to: //! connections and hands work to them — callers never touch a `Connection`.
//! - Switch between sync (rusqlite) and async (tokio-rusqlite) implementations //!
//! - Prevent blocking the async runtime with synchronous database calls //! - **Writes** (`execute`, `insert`, `transaction`, …) are sent as jobs to one
//! - Test with different database backends //! dedicated writer thread that owns the read-write connection. SQLite allows
//! - Migrate to other database systems in the future //! 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 crate::utils::lock::MutexSafe;
use async_trait::async_trait; use async_trait::async_trait;
use log::{debug, error};
use rusqlite::{params_from_iter, Connection, Result as SqliteResult, Row}; 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 /// Database query result type
pub type DbResult<T> = Result<T, String>; pub type DbResult<T> = Result<T, String>;
@@ -84,8 +99,28 @@ pub trait DatabaseService: Send + Sync {
F: FnOnce(&mut Transaction) -> DbResult<T> + Send + 'static, F: FnOnce(&mut Transaction) -> DbResult<T> + Send + 'static,
T: Send + 'static; T: Send + 'static;
/// Get the row ID of the most recent successful INSERT /// Run a transaction with foreign-key enforcement switched off for its
async fn last_insert_rowid(&self) -> DbResult<i64>; /// 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<F, T>(&self, f: F) -> DbResult<T>
where
F: FnOnce(&mut Transaction) -> DbResult<T> + 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<i64>;
/// 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 /// Transaction handle for batching multiple operations
@@ -110,48 +145,183 @@ impl<'a> Transaction<'a> {
} }
} }
/// Rusqlite-based database service implementation type Job = Box<dyn FnOnce(&Connection) + Send>;
///
/// This implementation wraps synchronous rusqlite operations in tokio::task::spawn_blocking /// The thread that owns the read-write connection. Jobs run one at a time, in
/// to prevent blocking the async runtime. /// 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<Job>,
}
impl Writer {
fn spawn(conn: Arc<Mutex<Connection>>) -> Self {
let (jobs, queue) = mpsc::channel::<Job>();
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<T, F>(&self, f: F) -> DbResult<T>
where
T: Send + 'static,
F: FnOnce(&Connection) -> DbResult<T> + 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<Vec<Connection>>,
returned: Condvar,
}
impl ReaderPool {
/// Blocking: waits for a free connection. Call from `spawn_blocking`.
fn run<T>(&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<Connection>,
}
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 { pub struct RusqliteService {
conn: Arc<Mutex<Connection>>, writer: Arc<Writer>,
readers: Option<Arc<ReaderPool>>,
} }
impl RusqliteService { 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<Mutex<Connection>>) -> Self { pub fn new(conn: Arc<Mutex<Connection>>) -> 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<Mutex<Connection>>, readers: Vec<Connection>) -> 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<T, F>(&self, f: F) -> DbResult<T>
where
T: Send + 'static,
F: FnOnce(&Connection) -> DbResult<T> + 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] #[async_trait]
impl DatabaseService for RusqliteService { impl DatabaseService for RusqliteService {
async fn execute(&self, query: Query) -> DbResult<usize> { async fn execute(&self, query: Query) -> DbResult<usize> {
let conn = Arc::clone(&self.conn); self.writer
tokio::task::spawn_blocking(move || { .run(move |conn| execute_query(conn, query))
// `lock_safe`, not `lock`: this is the busiest lock in the app and a .await
// 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))?
} }
async fn execute_batch(&self, sql: &str) -> DbResult<()> { async fn execute_batch(&self, sql: &str) -> DbResult<()> {
let conn = Arc::clone(&self.conn);
let sql = sql.to_string(); let sql = sql.to_string();
tokio::task::spawn_blocking(move || { self.writer
// `lock_safe`, not `lock`: this is the busiest lock in the app and a .run(move |conn| {
// panic under the guard would otherwise poison it, failing every conn.execute_batch(&sql)
// later query with "poisoned lock" until the process restarts. .map_err(|e| format!("Execute batch failed: {}", e))
let conn = conn.lock_safe(); })
conn.execute_batch(&sql) .await
.map_err(|e| format!("Execute batch failed: {}", e))
})
.await
.map_err(|e| format!("Task join error: {}", e))?
} }
async fn query_one<T, F>(&self, query: Query, mapper: F) -> DbResult<T> async fn query_one<T, F>(&self, query: Query, mapper: F) -> DbResult<T>
@@ -159,16 +329,7 @@ impl DatabaseService for RusqliteService {
T: Send + 'static, T: Send + 'static,
F: Fn(&Row) -> SqliteResult<T> + Send + 'static, F: Fn(&Row) -> SqliteResult<T> + Send + 'static,
{ {
let conn = Arc::clone(&self.conn); self.read(move |conn| query_one(conn, query, mapper)).await
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))?
} }
async fn query_optional<T, F>(&self, query: Query, mapper: F) -> DbResult<Option<T>> async fn query_optional<T, F>(&self, query: Query, mapper: F) -> DbResult<Option<T>>
@@ -176,16 +337,8 @@ impl DatabaseService for RusqliteService {
T: Send + 'static, T: Send + 'static,
F: Fn(&Row) -> SqliteResult<T> + Send + 'static, F: Fn(&Row) -> SqliteResult<T> + Send + 'static,
{ {
let conn = Arc::clone(&self.conn); self.read(move |conn| query_optional(conn, query, mapper))
tokio::task::spawn_blocking(move || { .await
// `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))?
} }
async fn query_many<T, F>(&self, query: Query, mapper: F) -> DbResult<Vec<T>> async fn query_many<T, F>(&self, query: Query, mapper: F) -> DbResult<Vec<T>>
@@ -193,16 +346,7 @@ impl DatabaseService for RusqliteService {
T: Send + 'static, T: Send + 'static,
F: Fn(&Row) -> SqliteResult<T> + Send + 'static, F: Fn(&Row) -> SqliteResult<T> + Send + 'static,
{ {
let conn = Arc::clone(&self.conn); self.read(move |conn| query_many(conn, query, mapper)).await
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))?
} }
async fn transaction<F, T>(&self, f: F) -> DbResult<T> async fn transaction<F, T>(&self, f: F) -> DbResult<T>
@@ -210,47 +354,65 @@ impl DatabaseService for RusqliteService {
F: FnOnce(&mut Transaction) -> DbResult<T> + Send + 'static, F: FnOnce(&mut Transaction) -> DbResult<T> + Send + 'static,
T: Send + 'static, T: Send + 'static,
{ {
let conn = Arc::clone(&self.conn); self.writer.run(move |conn| run_transaction(conn, f)).await
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))?
} }
async fn last_insert_rowid(&self) -> DbResult<i64> { async fn transaction_without_foreign_keys<F, T>(&self, f: F) -> DbResult<T>
let conn = Arc::clone(&self.conn); where
tokio::task::spawn_blocking(move || { F: FnOnce(&mut Transaction) -> DbResult<T> + Send + 'static,
// `lock_safe`, not `lock`: this is the busiest lock in the app and a T: Send + 'static,
// panic under the guard would otherwise poison it, failing every {
// later query with "poisoned lock" until the process restarts. self.writer
let conn = conn.lock_safe(); .run(move |conn| {
Ok(conn.last_insert_rowid()) conn.execute_batch("PRAGMA foreign_keys = OFF")
}) .map_err(|e| format!("Failed to disable foreign keys: {}", e))?;
.await let result = run_transaction(conn, f);
.map_err(|e| format!("Task join error: {}", e))? // 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<i64> {
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<F, T>(conn: &Connection, f: F) -> DbResult<T>
where
F: FnOnce(&mut Transaction) -> DbResult<T>,
{
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)
}
} }
} }
+327 -26
View File
@@ -17,9 +17,27 @@ use rusqlite::{Connection, Result as SqliteResult};
pub use db_service::{DatabaseService, RusqliteService}; pub use db_service::{DatabaseService, RusqliteService};
use schema::MIGRATIONS; 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 { 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<Mutex<Connection>>, conn: Arc<Mutex<Connection>>,
service: RusqliteService,
path: PathBuf, path: PathBuf,
} }
@@ -36,18 +54,48 @@ impl Database {
// Enable foreign keys // Enable foreign keys
conn.execute_batch("PRAGMA foreign_keys = ON;")?; 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;")?; 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 { let conn = Arc::new(Mutex::new(conn));
conn: Arc::new(Mutex::new(conn)), Self::migrate_connection(&conn, MIGRATIONS)?;
// Planner statistics. Without them SQLite guesses between indexes, and
// guessed badly for the listing query (see 08-database-design.md →
// "Listing query shape"). `optimize` only analyses what is missing or
// stale; `analysis_limit` bounds each table's scan so this stays in the
// milliseconds on a large catalogue. Failure is not fatal.
if let Err(e) = conn
.lock_safe()
.execute_batch("PRAGMA analysis_limit = 400; PRAGMA optimize = 0x10002;")
{
error!("PRAGMA optimize failed: {}", e);
}
// Readers open after migrations, so they only ever see the final schema.
let readers = (0..READER_CONNECTIONS)
.map(|_| Self::open_reader(path))
.collect::<SqliteResult<Vec<_>>>()?;
Ok(Self {
service: RusqliteService::with_readers(Arc::clone(&conn), readers),
conn,
path: path.clone(), path: path.clone(),
}; })
}
// Run migrations /// A connection that can only read. `query_only` makes an accidental write
db.migrate()?; /// routed to the pool fail loudly instead of racing the writer.
fn open_reader(path: &PathBuf) -> SqliteResult<Connection> {
Ok(db) 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) /// Open an in-memory database (for testing)
@@ -58,15 +106,16 @@ impl Database {
// Enable foreign keys // Enable foreign keys
conn.execute_batch("PRAGMA foreign_keys = ON;")?; conn.execute_batch("PRAGMA foreign_keys = ON;")?;
let db = Self { let conn = Arc::new(Mutex::new(conn));
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:"), path: PathBuf::from(":memory:"),
}; })
// Run migrations
db.migrate()?;
Ok(db)
} }
/// Get connection (for testing) /// Get connection (for testing)
@@ -75,11 +124,19 @@ impl Database {
Arc::clone(&self.conn) 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<()> { pub fn migrate(&self) -> SqliteResult<()> {
self.migrate_with(MIGRATIONS) 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. /// Apply `migrations` in order, skipping ones `_migrations` already records.
/// ///
/// **Each migration is one transaction, and the `_migrations` row is written /// **Each migration is one transaction, and the `_migrations` row is written
@@ -96,12 +153,16 @@ impl Database {
/// Every migration is pure DDL/DML, which SQLite runs transactionally — a /// 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. /// `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 /// TRACES: UR-002 | DR-012 | UT-014
fn migrate_with(&self, migrations: &[(&str, &str)]) -> SqliteResult<()> { fn migrate_connection(
conn: &Mutex<Connection>,
migrations: &[(&str, &str)],
) -> SqliteResult<()> {
info!("Starting database migrations..."); info!("Starting database migrations...");
let conn = self.conn.lock_safe(); let conn = conn.lock_safe();
// Create migrations table if it doesn't exist // Create migrations table if it doesn't exist
debug!("Creating _migrations table if it doesn't exist..."); debug!("Creating _migrations table if it doesn't exist...");
@@ -160,12 +221,10 @@ impl Database {
Ok(()) Ok(())
} }
/// Get a database service for async-safe operations /// A handle to the database service. Cheap: every call returns a clone of
/// /// the same writer thread and reader pool.
/// This wraps all blocking database operations in spawn_blocking to prevent
/// freezing the async runtime.
pub fn service(&self) -> RusqliteService { pub fn service(&self) -> RusqliteService {
RusqliteService::new(Arc::clone(&self.conn)) self.service.clone()
} }
/// Get the database file path /// Get the database file path
@@ -880,4 +939,246 @@ mod tests {
assert_eq!(user_id, "user2"); assert_eq!(user_id, "user2");
assert_eq!(username, "recent_user"); 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");
}
/// The planner gets statistics: the app used to never run `ANALYZE`, so
/// SQLite guessed between indexes — and for the listing query guessed the
/// `server_id` index, which every row shares, turning an index lookup into
/// a walk of the whole catalogue. `PRAGMA optimize` at open refreshes
/// whatever statistics are missing or stale, bounded by `analysis_limit`.
///
/// TRACES: UR-002 | DR-012 | UT-014
#[test]
fn open_gives_the_planner_statistics() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("jellytau.db");
{
let db = Database::open(&path).unwrap();
let conn = db.connection();
let conn = conn.lock_safe();
conn.execute_batch(
"INSERT INTO servers (id, name, url) VALUES ('s', 'S', 'http://s');",
)
.unwrap();
for i in 0..2000 {
conn.execute(
"INSERT INTO items (id, server_id, name, item_type) VALUES (?1, 's', 'n', 'Audio')",
[format!("i{i}")],
)
.unwrap();
}
}
let db = Database::open(&path).unwrap();
let conn = db.connection();
let conn = conn.lock_safe();
let analysed: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE name = 'sqlite_stat1'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(analysed, 1, "the planner has no statistics");
let items_stats: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_stat1 WHERE tbl = 'items'",
[],
|r| r.get(0),
)
.unwrap();
assert!(items_stats > 0, "no statistics for items");
}
/// 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<String> = 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();
}
} }
+164
View File
@@ -31,6 +31,7 @@ pub const MIGRATIONS: &[(&str, &str)] = &[
("024_multi_user_profiles", MIGRATION_024), ("024_multi_user_profiles", MIGRATION_024),
("025_backfill_item_library_id", MIGRATION_025), ("025_backfill_item_library_id", MIGRATION_025),
("026_server_catalog_generation", MIGRATION_026), ("026_server_catalog_generation", MIGRATION_026),
("027_items_container_id", MIGRATION_027),
]; ];
/// Initial schema migration /// Initial schema migration
@@ -941,6 +942,169 @@ const MIGRATION_026: &str = r#"
ALTER TABLE servers ADD COLUMN catalog_generation TEXT; ALTER TABLE servers ADD COLUMN catalog_generation TEXT;
"#; "#;
/// One canonical "which container lists this item" link, plus an index that
/// serves a listing in display order.
///
/// Jellyfin's `ParentId` is the *storage* parent, not the logical one: in a
/// series without season folders an episode's `ParentId` is the series while
/// its `SeasonId` names a virtual season, and a cached episode may arrive
/// without its season row at all. So listings matched children on four
/// columns at once (`parent_id`, `album_id`, `season_id`, `series_id`). That
/// was slow — the `OR` defeated the planner into walking the whole table — and
/// wrong: every episode carries its series id, so a series answered with its
/// seasons *and* all their episodes.
///
/// `container_id` resolves the logical container once, by rule: an episode
/// belongs to its season (else its series, else its parent), a season to its
/// series, a track to its album, anything else to its parent. It is a VIRTUAL
/// generated column, so every write path — cache, downloads, catalog crawl —
/// is covered without touching any of them, and it cannot drift from the
/// columns it is computed from. The index covers the listing's
/// `ORDER BY sort_name, name` (`sort_name` is usually NULL in the cache).
///
/// The placeholders keep offline navigation intact: an episode whose season
/// or series row was never cached used to surface directly under the series
/// through the `series_id` match. Now it lists under its season, so the season
/// (and series, and a track's album) must exist. They are built from the
/// names the child rows already carry, with `synced_at` NULL — they only show
/// when a download makes them available, and a real row from the server
/// replaces them wholesale (`save_to_cache` upserts every field).
///
/// TRACES: UR-002, UR-007 | DR-013
const MIGRATION_027: &str = r#"
ALTER TABLE items ADD COLUMN container_id TEXT GENERATED ALWAYS AS (
CASE item_type
WHEN 'Episode' THEN COALESCE(season_id, series_id, parent_id)
WHEN 'Season' THEN COALESCE(series_id, parent_id)
WHEN 'Audio' THEN COALESCE(album_id, parent_id)
ELSE parent_id
END
) VIRTUAL;
CREATE INDEX IF NOT EXISTS idx_items_container ON items(container_id, sort_name, name);
INSERT OR IGNORE INTO items (id, server_id, library_id, name, item_type, is_folder, series_id, series_name)
SELECT season_id, server_id, MAX(library_id), COALESCE(MAX(season_name), 'Season'), 'Season', 1,
MAX(series_id), MAX(series_name)
FROM items
WHERE item_type = 'Episode' AND season_id IS NOT NULL
GROUP BY season_id;
INSERT OR IGNORE INTO items (id, server_id, library_id, name, item_type, is_folder)
SELECT series_id, server_id, MAX(library_id), COALESCE(MAX(series_name), 'Series'), 'Series', 1
FROM items
WHERE item_type IN ('Episode', 'Season') AND series_id IS NOT NULL
GROUP BY series_id;
INSERT OR IGNORE INTO items (id, server_id, library_id, name, item_type, is_folder, album_artist)
SELECT album_id, server_id, MAX(library_id), COALESCE(MAX(album_name), 'Album'), 'MusicAlbum', 1,
MAX(album_artist)
FROM items
WHERE item_type = 'Audio' AND album_id IS NOT NULL
GROUP BY album_id;
"#;
#[cfg(test)]
mod migration_027_tests {
use super::*;
use rusqlite::Connection;
fn pre_027_db() -> Connection {
let conn = Connection::open_in_memory().unwrap();
let upto = MIGRATIONS
.iter()
.position(|(name, _)| *name == "027_items_container_id")
.expect("migration 027 must be registered");
for (_, sql) in &MIGRATIONS[..upto] {
conn.execute_batch(sql).unwrap();
}
conn.execute_batch(
"INSERT INTO servers (id, name, url) VALUES ('s', 'S', 'http://s');
-- An episode cached without its season or series rows.
INSERT INTO items (id, server_id, name, item_type, parent_id, season_id, season_name,
series_id, series_name, library_id)
VALUES ('ep', 's', 'Pilot', 'Episode', NULL, 'season', 'Season 1',
'show', 'Show', NULL);
-- A track cached without its album.
INSERT INTO items (id, server_id, name, item_type, album_id, album_name, album_artist)
VALUES ('trk', 's', 'Song', 'Audio', 'alb', 'Record', 'Band');
-- A folder child: its container is just its parent.
INSERT INTO items (id, server_id, name, item_type) VALUES ('box', 's', 'Box', 'BoxSet');
INSERT INTO items (id, server_id, name, item_type, parent_id)
VALUES ('film', 's', 'Film', 'Movie', 'box');",
)
.unwrap();
conn
}
fn container(conn: &Connection, id: &str) -> Option<String> {
conn.query_row("SELECT container_id FROM items WHERE id = ?1", [id], |r| {
r.get(0)
})
.unwrap()
}
/// TRACES: UR-002, UR-007 | DR-013
#[test]
fn every_item_resolves_to_its_logical_container() {
let conn = pre_027_db();
conn.execute_batch(MIGRATION_027).unwrap();
assert_eq!(container(&conn, "ep").as_deref(), Some("season"));
assert_eq!(container(&conn, "season").as_deref(), Some("show"));
assert_eq!(container(&conn, "trk").as_deref(), Some("alb"));
assert_eq!(container(&conn, "film").as_deref(), Some("box"));
assert_eq!(container(&conn, "show"), None);
}
/// Containers that were never cached get placeholders named from their
/// children, so an offline episode is still reachable series → season.
///
/// TRACES: UR-002, UR-007 | DR-013
#[test]
fn missing_containers_get_named_placeholders() {
let conn = pre_027_db();
conn.execute_batch(MIGRATION_027).unwrap();
let row = |id: &str| -> (String, String, Option<String>) {
conn.query_row(
"SELECT name, item_type, synced_at FROM items WHERE id = ?1",
[id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)
.unwrap()
};
assert_eq!(row("season"), ("Season 1".into(), "Season".into(), None));
assert_eq!(row("show"), ("Show".into(), "Series".into(), None));
assert_eq!(row("alb"), ("Record".into(), "MusicAlbum".into(), None));
}
/// Listing a container is one index range, already in display order.
///
/// TRACES: UR-002, UR-007 | DR-013
#[test]
fn a_container_listing_is_an_ordered_index_range() {
let conn = pre_027_db();
conn.execute_batch(MIGRATION_027).unwrap();
let plan: Vec<String> = conn
.prepare(
"EXPLAIN QUERY PLAN SELECT id FROM items
WHERE container_id = ?1 ORDER BY sort_name, name",
)
.unwrap()
.query_map(["season"], |r| r.get::<_, String>(3))
.unwrap()
.map(Result::unwrap)
.collect();
let plan = plan.join("\n");
assert!(plan.contains("idx_items_container"), "{plan}");
assert!(
!plan.contains("TEMP B-TREE"),
"listing needs a sort step:\n{plan}"
);
}
}
#[cfg(test)] #[cfg(test)]
mod migration_024_tests { mod migration_024_tests {
use super::*; use super::*;
+7 -10
View File
@@ -122,7 +122,7 @@ impl ThumbnailCache {
{ {
let path = PathBuf::from(&path_str); let path = PathBuf::from(&path_str);
if path.exists() { 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); return Some(path);
} }
// File gone — drop the stale row and fall through to the tag-agnostic // 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_str: String = db.query_optional(any_tag, |row| row.get(0)).await.ok()??;
let path = PathBuf::from(&path_str); let path = PathBuf::from(&path_str);
if path.exists() { if path.exists() {
self.touch(&db, item_id, image_type, None).await; self.touch(&db, item_id, image_type, None);
Some(path) Some(path)
} else { } else {
None None
@@ -166,13 +166,7 @@ impl ThumbnailCache {
/// Update `last_accessed` for LRU tracking. When `tag` is `Some`, scope to /// Update `last_accessed` for LRU tracking. When `tag` is `Some`, scope to
/// that exact row; when `None`, touch every row for the item + type. /// that exact row; when `None`, touch every row for the item + type.
async fn touch( fn touch(&self, db: &Arc<RusqliteService>, item_id: &str, image_type: &str, tag: Option<&str>) {
&self,
db: &Arc<RusqliteService>,
item_id: &str,
image_type: &str,
tag: Option<&str>,
) {
let query = match tag { let query = match tag {
Some(tag) => Query::with_params( Some(tag) => Query::with_params(
"UPDATE thumbnails SET last_accessed = CURRENT_TIMESTAMP "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 /// Save thumbnail to cache
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"$schema": "https://schema.tauri.app/config/2", "$schema": "https://schema.tauri.app/config/2",
"productName": "JellyTau", "productName": "JellyTau",
"version": "0.13.0", "version": "0.13.2",
"identifier": "com.dtourolle.jellytau", "identifier": "com.dtourolle.jellytau",
"build": { "build": {
"beforeDevCommand": "bun run dev", "beforeDevCommand": "bun run dev",
+43 -2
View File
@@ -66,9 +66,14 @@ async playerEnterBackgroundAudio(item: PlayItemRequest, positionSeconds: number)
return await TAURI_INVOKE("player_enter_background_audio", { item, positionSeconds }); return await TAURI_INVOKE("player_enter_background_audio", { item, positionSeconds });
}, },
/** /**
* TRACES: UR-040 | DR-052 | UT-061, IT-013 * Returns the item the native player is on and its absolute position. The
* item matters: an episode that ended while backgrounded has already advanced
* in the backend, so reloading the video the webview was mounted with would
* bring back the previous episode. (DR-296)
*
* TRACES: UR-040, UR-023 | DR-052, DR-296 | UT-061, IT-013
*/ */
async playerExitBackgroundAudio() : Promise<number> { async playerExitBackgroundAudio() : Promise<BackgroundAudioResume> {
return await TAURI_INVOKE("player_exit_background_audio"); return await TAURI_INVOKE("player_exit_background_audio");
}, },
/** /**
@@ -1590,6 +1595,16 @@ async repositoryGetSeriesEpisodes(handle: string, seriesId: string) : Promise<Me
async repositoryGetSeriesCurrentEpisode(handle: string, seriesId: string) : Promise<MediaItem | null> { async repositoryGetSeriesCurrentEpisode(handle: string, seriesId: string) : Promise<MediaItem | null> {
return await TAURI_INVOKE("repository_get_series_current_episode", { handle, seriesId }); return await TAURI_INVOKE("repository_get_series_current_episode", { handle, seriesId });
}, },
/**
* A series' episodes and the viewer's current episode, from one season
* fan-out. The series page used to ask for these as two commands, each of
* which walked every season.
*
* TRACES: UR-062 | DR-101, DR-295
*/
async repositoryGetSeriesView(handle: string, seriesId: string) : Promise<SeriesView> {
return await TAURI_INVOKE("repository_get_series_view", { handle, seriesId });
},
/** /**
* Erase the viewer's watch history for an item. * Erase the viewer's watch history for an item.
* *
@@ -2180,6 +2195,22 @@ export type BackgroundAction =
* Stop making sound. The user did not ask for background playback. * Stop making sound. The user did not ask for background playback.
*/ */
"pause" "pause"
/**
* Where playback stands when a background-audio handoff returns to the
* foreground. See [`PlayerController::background_audio_resume`].
*
* TRACES: UR-040, UR-023 | DR-296
*/
export type BackgroundAudioResume = {
/**
* Item the native audio player is on — `None` if the queue emptied (e.g.
* the sleep timer stopped playback while backgrounded).
*/
itemId: string | null;
/**
* Absolute position in that item, in seconds.
*/
positionSeconds: number }
/** /**
* Smart caching configuration * Smart caching configuration
*/ */
@@ -3445,6 +3476,16 @@ export type SecurityStatus = { usingKeyring: boolean; storageType: string }
* Audio track preference for a series * Audio track preference for a series
*/ */
export type SeriesAudioPreference = { seriesId: string; audioTrackDisplayTitle: string | null; audioTrackLanguage: string | null; audioTrackIndex: number | null } export type SeriesAudioPreference = { seriesId: string; audioTrackDisplayTitle: string | null; audioTrackLanguage: string | null; audioTrackIndex: number | null }
/**
* A series' episodes and the one the viewer is up to, from **one** season
* fan-out.
*
* The series page needs both, and asked for them as two commands; each walked
* every season, so every visit listed the show twice. One call, one walk.
*
* TRACES: UR-062 | DR-101, DR-295
*/
export type SeriesView = { episodes: MediaItem[]; current: MediaItem | null }
/** /**
* The verdict on a server's version. * The verdict on a server's version.
* *
+18 -1
View File
@@ -3,7 +3,13 @@
// NO direct HTTP calls - everything routes through Rust backend // NO direct HTTP calls - everything routes through Rust backend
import { commands } from "./bindings"; import { commands } from "./bindings";
import type { DownloadDiskUsage, JRayActor, SearchScope, StreamSelection } from "./bindings"; import type {
DownloadDiskUsage,
JRayActor,
SearchScope,
SeriesView,
StreamSelection,
} from "./bindings";
import type { QualityPreset } from "./quality-presets"; import type { QualityPreset } from "./quality-presets";
import type { import type {
Library, Library,
@@ -164,6 +170,17 @@ export class RepositoryClient {
return commands.repositoryGetSeriesCurrentEpisode(this.ensureHandle(), seriesId); return commands.repositoryGetSeriesCurrentEpisode(this.ensureHandle(), seriesId);
} }
/**
* A series' episodes and the episode the viewer is up to, from one season
* fan-out in Rust. Prefer this over calling `getSeriesEpisodes` and
* `getSeriesCurrentEpisode` together — each of those walks every season.
*
* TRACES: UR-062 | DR-101, DR-295
*/
async getSeriesView(seriesId: string): Promise<SeriesView> {
return commands.repositoryGetSeriesView(this.ensureHandle(), seriesId);
}
/** /**
* Erase watch history for an item. On a series or season the server applies * Erase watch history for an item. On a series or season the server applies
* it to everything inside, so the container returns to "never watched". * it to everything inside, so the container returns to "never watched".
+23 -3
View File
@@ -120,6 +120,12 @@
onNext?: () => void; // Called when user clicks next episode button onNext?: () => void; // Called when user clicks next episode button
hasNext?: boolean; // Whether there is a next episode available hasNext?: boolean; // Whether there is a next episode available
isLive?: boolean; // Live stream (Live TV) - no seek bar, no resume, no progress reporting isLive?: boolean; // Live stream (Live TV) - no seek bar, no resume, no progress reporting
/**
* Returning from background audio found the backend on a different item
* (the episode advanced while backgrounded). The page switches to it; this
* player must not reload the one it was mounted with. (DR-296)
*/
onResumeOtherItem?: (itemId: string, positionSeconds: number) => void;
} }
let { let {
@@ -137,6 +143,7 @@
onNext, onNext,
hasNext = false, hasNext = false,
isLive = false, isLive = false,
onResumeOtherItem,
}: Props = $props(); }: Props = $props();
// The id this player instance reports progress against. Snapshotted from the // The id this player instance reports progress against. Snapshotted from the
@@ -2035,9 +2042,12 @@
const wasPlaying = shouldResumeOnForeground(handoffState.wasPlaying, get(playerState).kind); const wasPlaying = shouldResumeOnForeground(handoffState.wasPlaying, get(playerState).kind);
handoffState = { ...initialHandoffState }; handoffState = { ...initialHandoffState };
try { try {
// Absolute position the native audio reached (base offset applied in Rust). // The item the native audio is on and the absolute position it reached
const pos = await commands.playerExitBackgroundAudio(); // (base offset applied in Rust). The item may not be `media`: an episode
log.debug("Returning from background audio at:", pos.toFixed(1)); // that ended while backgrounded has already advanced in the backend.
const resume = await commands.playerExitBackgroundAudio();
const pos = resume.positionSeconds;
log.debug("Returning from background audio at:", pos.toFixed(1), "item:", resume.itemId);
isMediaReady = false; isMediaReady = false;
// The foreground seek below (pendingForegroundSeek/handleCanPlay) OWNS the // The foreground seek below (pendingForegroundSeek/handleCanPlay) OWNS the
@@ -2056,8 +2066,18 @@
position: pos, position: pos,
wasPlaying, wasPlaying,
nativeStateKind: get(playerState).kind, nativeStateKind: get(playerState).kind,
mountedItemId: media?.id ?? null,
resumeItemId: resume.itemId,
}); });
// Reloading `media` here would bring back the previous episode at the new
// one's timestamp. Hand the switch to the page instead.
// TRACES: UR-040, UR-023 | DR-296
if (plan.target === "other-item" && plan.itemId) {
onResumeOtherItem?.(plan.itemId, plan.position);
return;
}
pendingForegroundPlay = plan.shouldPlay; pendingForegroundPlay = plan.shouldPlay;
// Determine the target stream + how the element/offset should be // Determine the target stream + how the element/offset should be
@@ -138,6 +138,51 @@ describe("backgroundAudioHandoff", () => {
}); });
expect(plan.position).toBe(0); expect(plan.position).toBe(0);
}); });
// An episode that ends while backgrounded advances in the backend. Reloading
// the video the player was mounted with brought back the PREVIOUS episode,
// at the new episode's timestamp.
//
// TRACES: UR-040, UR-023 | DR-296 | UT-265
it("switches to the item the backend advanced to", () => {
const plan = planHandoffReturn({
useHtml5Element: false,
position: 95,
wasPlaying: true,
nativeStateKind: "playing",
mountedItemId: "ep1",
resumeItemId: "ep2",
});
expect(plan.target).toBe("other-item");
expect(plan.itemId).toBe("ep2");
expect(plan.position).toBe(95);
});
it("reloads in place when the backend is still on the mounted item", () => {
const plan = planHandoffReturn({
useHtml5Element: true,
position: 95,
wasPlaying: true,
nativeStateKind: "playing",
mountedItemId: "ep1",
resumeItemId: "ep1",
});
expect(plan.target).toBe("html5-element");
});
it("reloads in place when the backend reports no item", () => {
// Queue emptied while backgrounded (e.g. the sleep timer): there is no
// other item to go to, so the mounted one is the best we have.
const plan = planHandoffReturn({
useHtml5Element: false,
position: 95,
wasPlaying: false,
nativeStateKind: undefined,
mountedItemId: "ep1",
resumeItemId: null,
});
expect(plan.target).toBe("native-backend");
});
}); });
}); });
@@ -77,8 +77,14 @@ export function shouldResumeOnForeground(
/** What has to be restarted to put picture back on screen, and how. */ /** What has to be restarted to put picture back on screen, and how. */
export interface HandoffReturn { export interface HandoffReturn {
/** Which renderer must be brought back. */ /**
target: "html5-element" | "native-backend"; * Which renderer must be brought back — or `other-item` when the backend is
* no longer on the item this player was mounted with, so the player must
* switch to `itemId` instead of reloading itself.
*/
target: "html5-element" | "native-backend" | "other-item";
/** The item to switch to; set only for `other-item`. */
itemId?: string;
/** Absolute position the background audio reached. */ /** Absolute position the background audio reached. */
position: number; position: number;
/** Whether playback should be running once it is back. */ /** Whether playback should be running once it is back. */
@@ -107,18 +113,31 @@ export interface HandoffReturn {
* `shouldPlay` folds in [shouldResumeOnForeground], so a lockscreen pause during * `shouldPlay` folds in [shouldResumeOnForeground], so a lockscreen pause during
* the handoff still wins over the snapshot taken on the way out. * the handoff still wins over the snapshot taken on the way out.
* *
* TRACES: UR-040, UR-003 | DR-196 | UT-060 * An episode that ends while backgrounded advances in the backend, so the item
* the native player returns on (`resumeItemId`) can differ from the one this
* player was mounted with. Reloading the mounted one brought back the previous
* episode at the new one's timestamp; in that case the plan is `other-item`.
* A missing `resumeItemId` (queue emptied) keeps the in-place reload.
*
* TRACES: UR-040, UR-003, UR-023 | DR-196, DR-296 | UT-060, UT-265
*/ */
export function planHandoffReturn(opts: { export function planHandoffReturn(opts: {
useHtml5Element: boolean; useHtml5Element: boolean;
position: number; position: number;
wasPlaying: boolean; wasPlaying: boolean;
nativeStateKind: string | undefined; nativeStateKind: string | undefined;
mountedItemId?: string | null;
resumeItemId?: string | null;
}): HandoffReturn { }): HandoffReturn {
const position = opts.position > 0 ? opts.position : 0;
const shouldPlay = shouldResumeOnForeground(opts.wasPlaying, opts.nativeStateKind);
if (opts.resumeItemId && opts.resumeItemId !== opts.mountedItemId) {
return { target: "other-item", itemId: opts.resumeItemId, position, shouldPlay };
}
return { return {
target: opts.useHtml5Element ? "html5-element" : "native-backend", target: opts.useHtml5Element ? "html5-element" : "native-backend",
position: opts.position > 0 ? opts.position : 0, position,
shouldPlay: shouldResumeOnForeground(opts.wasPlaying, opts.nativeStateKind), shouldPlay,
}; };
} }
@@ -32,12 +32,12 @@ describe("background-audio player commands (param naming)", () => {
}); });
}); });
it("player_exit_background_audio takes no params and returns a position", async () => { it("player_exit_background_audio takes no params and returns the resume point", async () => {
(invoke as any).mockResolvedValueOnce(193.5); (invoke as any).mockResolvedValueOnce({ itemId: "ep2", positionSeconds: 193.5 });
const pos = await commands.playerExitBackgroundAudio(); const resume = await commands.playerExitBackgroundAudio();
expect(pos).toBe(193.5); expect(resume).toEqual({ itemId: "ep2", positionSeconds: 193.5 });
expect(invoke).toHaveBeenCalledWith("player_exit_background_audio"); expect(invoke).toHaveBeenCalledWith("player_exit_background_audio");
}); });
}); });
+86
View File
@@ -0,0 +1,86 @@
import { describe, it, expect, vi } from "vitest";
import { createCoalescedLoader } from "./coalescedLoader";
/**
* TRACES: UR-062 | DR-295
*
* The series page loaded itself six times on every open: `onMount` and a
* `$effect` both ran on mount, the "server became reachable" effect fired on
* its first run, and navigation updates re-ran the effect. Each load repeated
* the item, the season list and the whole series view — about six times a
* dozen requests in flight at once, which alone slowed every server call on a
* phone to 2-3 s.
*/
function deferred() {
let resolve!: () => void;
const promise = new Promise<void>((r) => (resolve = r));
return { promise, resolve };
}
describe("createCoalescedLoader", () => {
it("shares one run between calls for the same key while it is in flight", async () => {
const gate = deferred();
const run = vi.fn(() => gate.promise);
const loader = createCoalescedLoader(run);
const calls = [1, 2, 3, 4, 5, 6].map(() => loader.load("frasier"));
gate.resolve();
await Promise.all(calls);
expect(run).toHaveBeenCalledTimes(1);
});
it("re-runs once after the in-flight load when a caller needs fresh data", async () => {
const gates = [deferred(), deferred()];
let n = 0;
const run = vi.fn(() => gates[n++].promise);
const loader = createCoalescedLoader(run);
const first = loader.load("frasier");
// e.g. "mark watched" finished while the page was still loading: the
// in-flight load may predate the change, so it must not be the answer.
const fresh = loader.load("frasier", { fresh: true });
const fresh2 = loader.load("frasier", { fresh: true });
gates[0].resolve();
await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(2));
gates[1].resolve();
// Every caller is answered by the load that includes the re-run.
await Promise.all([first, fresh, fresh2]);
expect(run).toHaveBeenCalledTimes(2);
});
it("does not share a run between different keys", async () => {
const run = vi.fn(() => Promise.resolve());
const loader = createCoalescedLoader(run);
await Promise.all([loader.load("frasier"), loader.load("cheers")]);
expect(run).toHaveBeenCalledTimes(2);
expect(run).toHaveBeenNthCalledWith(1, "frasier");
expect(run).toHaveBeenNthCalledWith(2, "cheers");
});
it("runs again once the previous load has finished", async () => {
const run = vi.fn(() => Promise.resolve());
const loader = createCoalescedLoader(run);
await loader.load("frasier");
await loader.load("frasier");
expect(run).toHaveBeenCalledTimes(2);
});
it("releases the key when a load fails", async () => {
const run = vi
.fn<(key: string) => Promise<void>>()
.mockRejectedValueOnce(new Error("offline"))
.mockResolvedValueOnce(undefined);
const loader = createCoalescedLoader(run);
await expect(loader.load("frasier")).rejects.toThrow("offline");
await loader.load("frasier");
expect(run).toHaveBeenCalledTimes(2);
});
});
+55
View File
@@ -0,0 +1,55 @@
/**
* Load one keyed thing at a time, however many triggers ask for it.
*
* Calls for the key already loading share that load instead of starting their
* own. A caller that knows the data changed (`fresh` — after "mark watched",
* on reconnect, when a filter flips) must not be answered by a load that may
* predate the change, so it gets exactly one re-run once the current load
* ends, however many such callers there were.
*
* Exists because the series page loaded itself six times on every open —
* `onMount`, a mount-time `$effect`, the reachability effect's first run and
* navigation updates each started a full load — putting about six times a
* dozen requests in flight at once.
*
* TRACES: UR-062 | DR-295
*/
export interface CoalescedLoader {
/** Load `key`. `fresh`: the caller knows the data changed. */
load(key: string, options?: { fresh?: boolean }): Promise<void>;
}
interface InFlight {
key: string;
/** Settles when this load and any re-run it owes have finished. */
done: Promise<void>;
rerun: boolean;
}
export function createCoalescedLoader(run: (key: string) => Promise<void>): CoalescedLoader {
let inFlight: InFlight | null = null;
return {
load(key, options = {}) {
if (inFlight && inFlight.key === key) {
if (options.fresh) inFlight.rerun = true;
return inFlight.done;
}
const entry: InFlight = { key, rerun: false, done: Promise.resolve() };
entry.done = (async () => {
try {
await run(key);
while (entry.rerun) {
entry.rerun = false;
await run(key);
}
} finally {
if (inFlight === entry) inFlight = null;
}
})();
inFlight = entry;
return entry.done;
},
};
}
+24 -23
View File
@@ -73,7 +73,8 @@
// a new page inherits the previous page's offset. Must be registered here at // a new page inherits the previous page's offset. Must be registered here at
// init, alongside the tracker above, for the same reason. (DR-156) // init, alongside the tracker above, for the same reason. (DR-156)
let shellScroller = $state<HTMLElement>(); let shellScroller = $state<HTMLElement>();
useScrollRestore(() => shellScroller, "shell"); // Owned routes scroll inside their own column; the shell box does not.
useScrollRestore(() => (routeOwnsLayout ? undefined : shellScroller), "shell");
// Layout-shell visibility rules live in one pure, unit-tested module // Layout-shell visibility rules live in one pure, unit-tested module
// ($lib/utils/layoutShell) so they can't drift per route/platform. // ($lib/utils/layoutShell) so they can't drift per route/platform.
@@ -357,29 +358,29 @@
scrolling internally. All other top-level pages render directly here, so scrolling internally. All other top-level pages render directly here, so
this wrapper must scroll and reserve the fixed bottom UI's measured this wrapper must scroll and reserve the fixed bottom UI's measured
height so the mini player / bottom nav never overlap the last rows. --> height so the mini player / bottom nav never overlap the last rows. -->
{#if routeOwnsLayout} <!-- Shared header (account menu, desktop nav) as a flex-shrink-0 sibling
<!-- These routes own their own full-height flex column (header + scroller above the scroller, so it never eats into the scroller's bounds. -->
+ their own in-flow BottomUi), so the root just clips and steps back. --> {#if !routeOwnsLayout && showGlobalHeader}
<div class="flex-1 overflow-hidden"> <AppHeader />
{@render children()}
</div>
{:else}
<!-- Shared header (account menu, desktop nav) as a flex-shrink-0 sibling
above the scroller, so it never eats into the scroller's bounds. -->
{#if showGlobalHeader}
<AppHeader />
{/if}
<!-- Scroller is flex-1/min-h-0; the in-flow BottomUi below is a flex
sibling, so the list is physically bounded above it and can never
render behind it. No measurement, no reserved padding. -->
<div
bind:this={shellScroller}
class="flex-1 overflow-y-auto min-h-0"
style="overscroll-behavior: contain"
>
{@render children()}
</div>
{/if} {/if}
<!-- ONE element renders the route, whatever the layout mode; only its
classes change. Routes that own their full-height column (header +
scroller + their own in-flow BottomUi) get a plain clipped box; every
other route gets the shell scroller (flex-1/min-h-0, bounded above the
in-flow BottomUi, so no measurement or reserved padding).
This used to be two branches, each rendering `children`. The page
store that decides the mode can update a flush after the new route
renders, so navigating between the two kinds of route (Search → a
library page) mounted the page under one branch and then *remounted*
it under the other — every load it started, twice (DR-295). -->
<div
bind:this={shellScroller}
class={routeOwnsLayout ? "flex-1 overflow-hidden" : "flex-1 overflow-y-auto min-h-0"}
style={routeOwnsLayout ? undefined : "overscroll-behavior: contain"}
>
{@render children()}
</div>
<!-- Re-authentication modal --> <!-- Re-authentication modal -->
<ReauthModal isOpen={$needsReauth} /> <ReauthModal isOpen={$needsReauth} />
+28 -23
View File
@@ -1,6 +1,6 @@
<!-- TRACES: UR-035, UR-038, UR-048, UR-058, UR-062 | DR-043, DR-062, DR-102, DR-103, DR-142 --> <!-- TRACES: UR-035, UR-038, UR-048, UR-058, UR-062 | DR-043, DR-062, DR-102, DR-103, DR-142 -->
<script lang="ts"> <script lang="ts">
import { onMount, untrack } from "svelte"; import { untrack } from "svelte";
import { formatDuration } from "$lib/utils/duration"; import { formatDuration } from "$lib/utils/duration";
import { page } from "$app/stores"; import { page } from "$app/stores";
import { goto } from "$app/navigation"; import { goto } from "$app/navigation";
@@ -51,6 +51,7 @@
type SeasonData, type SeasonData,
} from "$lib/components/library/seriesNavigation"; } from "$lib/components/library/seriesNavigation";
import { createLogger } from "$lib/utils/logger"; import { createLogger } from "$lib/utils/logger";
import { createCoalescedLoader } from "$lib/utils/coalescedLoader";
const log = createLogger("LibraryDetail"); const log = createLogger("LibraryDetail");
@@ -65,18 +66,27 @@
// preference, so it resets with each load (DR-107). // preference, so it resets with each load (DR-107).
let expandedSeasons = $state<Set<string>>(new Set()); let expandedSeasons = $state<Set<string>>(new Set());
// Track if we've done an initial load and previous server state // Track if we've done an initial load and previous server state. The
// previous state starts as "unknown" (null): the effect's first run only
// records it. Starting at `false` made that first run look like a
// reconnect and force a second, fresh load of the page on every open.
let hasLoadedOnce = false; let hasLoadedOnce = false;
let previousServerReachable = false; let previousServerReachable: boolean | null = null;
const itemId = $derived($page.params.id); const itemId = $derived($page.params.id);
const focusedEpisodeId = $derived($page.url.searchParams.get("episode")); const focusedEpisodeId = $derived($page.url.searchParams.get("episode"));
onMount(async () => { // Every trigger below goes through one coalesced loader: they used to each
await loadItem(); // start a full load, so opening a page loaded it six times over (DR-295).
hasLoadedOnce = true; // `fresh` marks triggers that know the data changed.
}); const loader = createCoalescedLoader(() => loadItemNow());
function loadItem(options?: { fresh?: boolean }): Promise<void> {
if (!itemId) return Promise.resolve();
return loader.load(itemId, options);
}
const reloadFresh = () => loadItem({ fresh: true });
// Runs on mount and whenever the item changes.
$effect(() => { $effect(() => {
if (itemId) { if (itemId) {
loadItem(); loadItem();
@@ -89,8 +99,8 @@
const serverReachable = $isServerReachable; const serverReachable = $isServerReachable;
// If server just became reachable and we've already loaded, reload to get fresh data // If server just became reachable and we've already loaded, reload to get fresh data
if (serverReachable && !previousServerReachable && hasLoadedOnce && itemId) { if (serverReachable && previousServerReachable === false && hasLoadedOnce && itemId) {
loadItem(); reloadFresh();
} }
previousServerReachable = serverReachable; previousServerReachable = serverReachable;
@@ -100,10 +110,10 @@
// contents follow the filter the same way a library listing does. // contents follow the filter the same way a library listing does.
// TRACES: UR-052 | DR-143 // TRACES: UR-052 | DR-143
useOfflineFilterReload(() => { useOfflineFilterReload(() => {
if (itemId) loadItem(); if (itemId) reloadFresh();
}); });
async function loadItem() { async function loadItemNow() {
if (!itemId) return; if (!itemId) return;
// Only show spinner when navigating to a different item // Only show spinner when navigating to a different item
// untrack prevents $effect from tracking `item` as a dependency (avoids infinite loop) // untrack prevents $effect from tracking `item` as a dependency (avoids infinite loop)
@@ -211,14 +221,9 @@
const repo = auth.getRepository(); const repo = auth.getRepository();
const seasons = $libraryItems.filter((i) => i.kind === "season"); const seasons = $libraryItems.filter((i) => i.kind === "season");
const [episodes, current] = await Promise.all([ // One call, one season fan-out: asking for episodes and the current
repo.getSeriesEpisodes(itemId), // episode separately walked every season twice (DR-295).
// Best-effort: a series still renders if the anchor cannot be resolved. const { episodes, current } = await repo.getSeriesView(itemId);
repo.getSeriesCurrentEpisode(itemId).catch((e) => {
log.warn("Could not resolve the current episode:", e);
return null;
}),
]);
seasonData = groupEpisodesBySeason(seasons, episodes); seasonData = groupEpisodesBySeason(seasons, episodes);
currentEpisode = current; currentEpisode = current;
@@ -584,13 +589,13 @@
watched={allEpisodes.length > 0 && allEpisodes.every((e) => e.userData?.isPlayed)} watched={allEpisodes.length > 0 && allEpisodes.every((e) => e.userData?.isPlayed)}
scope="series" scope="series"
showLabel={true} showLabel={true}
onChanged={loadItem} onChanged={reloadFresh}
/> />
<ClearHistoryButton <ClearHistoryButton
itemId={item.id} itemId={item.id}
itemName={item.name} itemName={item.name}
scope="series" scope="series"
onCleared={loadItem} onCleared={reloadFresh}
/> />
{:else if item.kind === "movie"} {:else if item.kind === "movie"}
<VideoDownloadButton <VideoDownloadButton
@@ -605,7 +610,7 @@
watched={item.userData?.isPlayed ?? false} watched={item.userData?.isPlayed ?? false}
scope="episode" scope="episode"
showLabel={true} showLabel={true}
onChanged={loadItem} onChanged={reloadFresh}
/> />
{/if} {/if}
<!-- Favourite. Sits with Play/Download rather than in the header, <!-- Favourite. Sits with Play/Download rather than in the header,
@@ -737,7 +742,7 @@
expanded={expandedSeasons.has(season.id)} expanded={expandedSeasons.has(season.id)}
onToggle={() => toggleSeason(season.id)} onToggle={() => toggleSeason(season.id)}
onEpisodeClick={handleEpisodeClick} onEpisodeClick={handleEpisodeClick}
onHistoryCleared={loadItem} onHistoryCleared={reloadFresh}
/> />
{/each} {/each}
{/if} {/if}
+27 -1
View File
@@ -50,6 +50,9 @@
// When advancing to a next episode we always start from the beginning, // When advancing to a next episode we always start from the beginning,
// even if the episode was previously started or watched. // even if the episode was previously started or watched.
const restartParam = $derived($page.url.searchParams.get("restart") === "true"); const restartParam = $derived($page.url.searchParams.get("restart") === "true");
// Explicit start position in seconds — set when returning from background
// audio onto an episode the backend advanced to (DR-296).
const resumeAtParam = $derived(Number($page.url.searchParams.get("resumeAt")) || 0);
// Derive playback context from URL query params // Derive playback context from URL query params
const playbackContext = $derived.by(() => { const playbackContext = $derived.by(() => {
@@ -121,6 +124,7 @@
$effect(() => { $effect(() => {
const id = itemId; const id = itemId;
const restart = restartParam; const restart = restartParam;
const resumeAt = resumeAtParam;
if (id && id !== loadedItemId) { if (id && id !== loadedItemId) {
autoPlayLog.debug( autoPlayLog.debug(
"$effect triggered: loading new item", "$effect triggered: loading new item",
@@ -132,7 +136,11 @@
); );
// restart=true (advancing to next episode) forces start-from-beginning, // restart=true (advancing to next episode) forces start-from-beginning,
// bypassing the resume-progress check. // bypassing the resume-progress check.
loadAndPlay(id, restart ? 0 : undefined, restart); if (resumeAt > 0) {
loadAndPlay(id, resumeAt);
} else {
loadAndPlay(id, restart ? 0 : undefined, restart);
}
} }
}); });
@@ -797,6 +805,23 @@
} }
} }
/**
* Background audio advanced to another episode; show that one where the audio
* left off, rather than the episode this page was opened on.
*
* The outgoing episode was played out (the backend advances only past its
* end), so it is recorded as watched and the VideoPlayer's unmount stop
* report — which would carry the stale handoff position — is suppressed, as
* for a manual skip.
*
* TRACES: UR-040, UR-023 | DR-296
*/
function handleResumeOtherItem(nextId: string, positionSeconds: number) {
void reportSkippedEpisode(currentMedia?.id ?? itemId ?? null);
const start = positionSeconds > 0 ? `resumeAt=${Math.floor(positionSeconds)}` : "restart=true";
goto(`/player/${nextId}?${start}`, { replaceState: true });
}
function handleSkipToNextEpisode() { function handleSkipToNextEpisode() {
if (nextEpisode) { if (nextEpisode) {
// Skipping means "I'm done with this one" — record the outgoing episode as // Skipping means "I'm done with this one" — record the outgoing episode as
@@ -889,6 +914,7 @@
onEnded={handleVideoEnded} onEnded={handleVideoEnded}
hasNext={nextEpisode !== null} hasNext={nextEpisode !== null}
onNext={handleSkipToNextEpisode} onNext={handleSkipToNextEpisode}
onResumeOtherItem={handleResumeOtherItem}
/> />
<NextEpisodePopup /> <NextEpisodePopup />
{:else} {:else}