diff --git a/docs/rfcs/RFC-016-audio-quality-analysis.md b/docs/rfcs/RFC-016-audio-quality-analysis.md index 64fe8a17..239c1fdd 100644 --- a/docs/rfcs/RFC-016-audio-quality-analysis.md +++ b/docs/rfcs/RFC-016-audio-quality-analysis.md @@ -6,12 +6,21 @@ (steps 2 to 4 together: the migration, the ported detector and the FFmpeg pipe, the background analyser, the verdicts, the five routes, the "Needs attention" category, the Quality page and the track properties' "Real - quality" section). The *Status* field above never flips in this project; + quality" section), then [#320](https://github.com/InstaZDLL/waveflow-server/pull/320) + (the analysis state in the summary and on the page, and one write per + measurement). The *Status* field above never flips in this project; this line is the one that says what is running. - **Date**: 2026-10-24 - **Revised**: 2026-10-25, with the implementation. `upsampled` stops at 30 kHz rather than 24, as measured on FFmpeg's own resampler (decision 3), and decision 3 says which upsampled files it cannot see. +- **Revised**: 2026-10-24, after the page ran on a real library and said + nothing for forty seconds at a time. The summary carries the analyser's + state (the surface), the page shows it (decision 10), and each + measurement is written as it finishes rather than sixteen at a time + (decision 4). A scan the caller cannot see changes nothing they are + shown, the header and the list refresh together, "Analyse again" is + offered only at rest, and each row shows its cover (decision 10). - **Authors**: the WaveFlow project - **Depends on**: [RFC-002](RFC-002-waveflow-server-v2.md) (FFmpeg, read-only audio, the path guard, tenancy in the queries), @@ -352,9 +361,15 @@ that cannot be read is not retried on every wake, which would loop forever on one bad file; it is retried when its content changes or the detector's version moves. -**Writes.** Results are buffered and written in batches of 16 rows in one -transaction, under the writer gate, as every mutation is. A batch never holds -the gate while FFmpeg runs. +**Writes.** Each measurement is written as it finishes, in a transaction of +its own, under the writer gate, as every mutation is. The gate is never held +while FFmpeg runs. The first version wrote sixteen rows at a time, which on +a debug build left the page's count still for forty seconds and then +jumping. The count is what a reader watches, and a row a file costs one +short transaction against a decode a hundred times longer. A restart loses +at most the files in flight, and a row that cannot be written costs that +file and not the pass. Decision 5's guarantees are unchanged: a row is still +written only while the track holds the content it measured. **Paths.** Every file goes through `secure_track_path` before FFmpeg sees it, like a stream does. Audio files stay read-only: FFmpeg reads from the path @@ -563,8 +578,18 @@ Folders page already uses. - **A header** gives the library's counts per verdict, as chips that filter the list, and the progress of the analysis: "1,204 of 3,310 lossless files - analysed". While anything is `pending` and the analysis is on, it refreshes - its counts every ten seconds; when nothing is, it stops. + analysed". +- **A status line** says what the analysis is doing, from the summary's + `analysis`: running, with the count moving and the files in progress; + waiting for the scan to finish; queued; or idle, with when the last file + was analysed. It is a polite live region, so a screen reader hears the + state change without being interrupted on every file. While the analysis + works (running, queued or waiting), the page asks again every three + seconds; idle or off, it stops asking. The header and the list are one + answer: both are asked again on every visit, and a change of filter asks + for the header beside the new list, so the counts above never contradict + the rows below. The first version held both for the client's thirty + seconds of cache, and a filter fetched a list fresher than its header. - **The list** is the song table the other pages use, with three columns of its own: the container (`FLAC · 24 bit · 44.1 kHz · 1,021 kbps`, from `file-quality.tsx`), where the highs stop (`16.2 kHz`, or "full band"), and @@ -576,7 +601,12 @@ Folders page already uses. the verdict in words, the numbers behind it, the averaged spectrum, and a "Show spectrogram" button. - **For an owner or manager**, an "Analyse again" action on the library, with - the second click RFC-015 already uses to confirm. + the second click RFC-015 already uses to confirm. It is offered only when + the server, asked since the page was opened, says the analysis is idle: + whether a pass runs is the server's to say, not something the page + remembers, and a page left and opened again remembers nothing. +- **Each row shows the track's cover** before its title, as the song tables + do: a title alone is hard to place. - **The meter in `file-quality.tsx` does not change.** It grades the container, which is what it says it does. The track's properties put the measured verdict under it, so both are visible and neither pretends to be @@ -588,12 +618,32 @@ All under `/api/v2`, tenancy-checked in the queries. | Route | What | | --- | --- | -| `GET /libraries/{id}/quality` | `{ enabled, lossless, measured, verdicts: [{ verdict, count }] }`. Any member. | +| `GET /libraries/{id}/quality` | `{ enabled, lossless, measured, verdicts: [{ verdict, count }], analysis }`. Any member. `analysis` is `{ state, in_flight, pass_started_at, last_measured_at }`, below. | | `GET /libraries/{id}/quality/tracks?verdict=&codec=&offset=&limit=` | The song items of `GET /libraries/{id}/tracks`, each with a `quality` object: `verdict`, `cutoff_hz`, `wall_db`, `nyquist_hz`, `effective_bits`, `analyzed_at`. Sorted as decision 10 says. `422` for an unknown verdict. Any member. | | `POST /libraries/{id}/quality/analysis` | Decision 5. `202`. Owner or manager; `404` for anyone else. | | `GET /tracks/{track_id}/quality` | One track's `quality` object, plus `spectrum`: the 256 bands as numbers in dB. `pending` with no numbers when nothing is current. Any member. | | `GET /tracks/{track_id}/spectrogram` | Decision 8. `image/webp`, or `image/jpeg` without libwebp. Any member; `429` when the bound is reached. | +**`analysis.state`** is one of: + +| State | When | +| --- | --- | +| `off` | The operator set `WAVEFLOW_QUALITY_PARALLELISM=0`. | +| `idle` | Nothing of this library is pending, and none of its files is being decoded. | +| `waiting_for_scan` | Files are pending, and a scan of a library the caller belongs to holds the analysis back. | +| `running` | Files are pending and a pass is running, including one held back by a scan the caller cannot see. | +| `queued` | Files are pending and no pass is running yet: the analyser is between passes. | + +A scan holds every measurement back, of every library, but only a scan the +caller can see is named. A scan of somebody else's library is not something +to confirm to them, so it changes nothing they are shown: the pass still +reads as `running`, and the file it picked as in flight, exactly as with no +scan at all. A file counts as in flight from the moment the pass picks it, +before any wait, for the same reason. `in_flight` counts this library's +files picked and not yet written, `pass_started_at` is set only while +`running`, and `last_measured_at` is the latest `analyzed_at` among this +library's current measurements. + Every read goes through `DomainServices`, so the desktop can reach the same answers when it chooses to, and so can a future Subsonic extension if one is ever specified. diff --git a/src/lib.rs b/src/lib.rs index f03534a0..7cce953a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -338,6 +338,8 @@ fn chunk_body_limit(limits: &config::UploadLimits) -> usize { services::CreditDecisionKind, services::CreditDecisionRecord, services::QualitySummary, + services::QualityAnalysis, + services::AnalysisState, services::QualityVerdict, services::QualityVerdictCount, services::QualityTrack, diff --git a/src/scanner.rs b/src/scanner.rs index 1d64ea53..70d1edb0 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -89,14 +89,27 @@ pub struct ScanManager { /// (RFC-016 decision 4) starts no file while this is above zero, and wakes /// when it falls back to it. in_flight: Arc>, + /// Which libraries those scans belong to, with how many of each. The + /// quality page says "waiting for the scan" only to someone who can see + /// the library being scanned. + scanning: Arc>, } /// One scan counted as running, for as long as this lives. -pub struct ScanInFlight(Arc>); +pub struct ScanInFlight { + count: Arc>, + scanning: Arc>, + library_id: Uuid, +} impl Drop for ScanInFlight { fn drop(&mut self) { - self.0.send_modify(|count| *count = count.saturating_sub(1)); + self.scanning.remove_if_mut(&self.library_id, |_, count| { + *count = count.saturating_sub(1); + *count == 0 + }); + self.count + .send_modify(|count| *count = count.saturating_sub(1)); } } @@ -133,14 +146,25 @@ impl ScanManager { events: Arc::new(DashMap::new()), library_locks: Arc::new(DashMap::new()), in_flight: Arc::new(watch::Sender::new(0)), + scanning: Arc::new(DashMap::new()), } } - /// Counts a scan as running until the guard is dropped. Every job takes - /// one for its whole run, success or failure. - pub fn begin_scan(&self) -> ScanInFlight { + /// Counts a scan of `library_id` as running until the guard is dropped. + /// Every job takes one for its whole run, success or failure. + pub fn begin_scan(&self, library_id: Uuid) -> ScanInFlight { + *self.scanning.entry(library_id).or_insert(0) += 1; self.in_flight.send_modify(|count| *count += 1); - ScanInFlight(Arc::clone(&self.in_flight)) + ScanInFlight { + count: Arc::clone(&self.in_flight), + scanning: Arc::clone(&self.scanning), + library_id, + } + } + + /// Whether a scan of this library is running. + pub fn is_scanning(&self, library_id: Uuid) -> bool { + self.scanning.contains_key(&library_id) } /// How many scans are running, and a way to wait for that to change. @@ -172,7 +196,7 @@ impl ScanManager { let (sender, _) = broadcast::channel(128); self.events.insert(scan_id, sender); let manager = self.clone(); - let running = self.begin_scan(); + let running = self.begin_scan(library_id); tokio::spawn(async move { let _running = running; if let Err(error) = manager.run(scan_id, library).await { diff --git a/src/services/mod.rs b/src/services/mod.rs index 97d704b6..6276c40a 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -1467,8 +1467,8 @@ pub use library_folders::{ pub use library_tree::{DiskFolder, DiskListing, DiskTrack}; pub use lyrics_uploads::{LyricsFile, MAX_UPLOAD_BYTES as MAX_LYRICS_UPLOAD_BYTES}; pub use quality::{ - QualitySummary, QualityTrack, QualityVerdict, QualityVerdictCount, Spectrogram, TrackQuality, - TrackQualityDetail, + AnalysisState, QualityAnalysis, QualitySummary, QualityTrack, QualityVerdict, + QualityVerdictCount, Spectrogram, TrackQuality, TrackQualityDetail, }; pub use scrobbling::{ ScrobbleDestinationName, ScrobbleDrain, ScrobbleEnvelope, ScrobbleLinkState, ScrobbleProvider, diff --git a/src/services/quality.rs b/src/services/quality.rs index 79bd32a9..64c8be74 100644 --- a/src/services/quality.rs +++ b/src/services/quality.rs @@ -19,7 +19,10 @@ //! query here call a row current on its `full_hash` alone. use std::{ - sync::{Arc, RwLock}, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, RwLock, + }, time::Duration, }; @@ -172,6 +175,39 @@ pub struct QualitySummary { pub measured: i64, /// Every verdict, worst first, zero counts included. pub verdicts: Vec, + /// Where the analysis stands for this library. + pub analysis: QualityAnalysis, +} + +/// What the analyser is doing about one library's files. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum AnalysisState { + /// The operator turned the analysis off. + Off, + /// Nothing of this library is waiting, and nothing is being measured. + Idle, + /// Files are waiting and no pass is running yet: the analyser is between + /// passes, and the next wake reaches them. + Queued, + /// Files are waiting, and a scan the caller can see holds the analysis + /// back until it finishes. + WaitingForScan, + /// A pass is running. One held back by a scan the caller cannot see is + /// still `running` to them: that scan is not theirs to know about. + Running, +} + +/// The analyser's state, as one library's member may read it. +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct QualityAnalysis { + pub state: AnalysisState, + /// How many of this library's files are being decoded right now. + pub in_flight: i64, + /// When the running pass began; `null` unless one is running. + pub pass_started_at: Option, + /// When a file of this library was last measured; `null` before any was. + pub last_measured_at: Option, } /// One file's verdict and the numbers behind it. @@ -235,6 +271,18 @@ struct QualityInner { spectrograms: Arc, /// One spectrogram per account at a time. spectrogram_users: dashmap::DashMap>, + /// When the running pass began, if one is running, and when the last one + /// ended. + pass: Mutex, + /// How many measurements are held back by a running scan. + waiting: AtomicUsize, + /// How many files are being decoded right now, by library. + measuring: dashmap::DashMap, +} + +#[derive(Default)] +struct PassState { + started_at: Option, } impl QualityRuntime { @@ -246,9 +294,19 @@ impl QualityRuntime { wake: Notify::new(), spectrograms: Arc::new(Semaphore::new(parallelism + 1)), spectrogram_users: dashmap::DashMap::new(), + pass: Mutex::new(PassState::default()), + waiting: AtomicUsize::new(0), + measuring: dashmap::DashMap::new(), })) } + fn pass(&self) -> std::sync::MutexGuard<'_, PassState> { + self.0 + .pass + .lock() + .unwrap_or_else(|poison| poison.into_inner()) + } + fn decoder(&self) -> Option { self.0 .decoder @@ -258,9 +316,61 @@ impl QualityRuntime { } } +/// Marks a pass as running for as long as it lives. +struct Passing(QualityRuntime); + +impl Passing { + fn begin(runtime: &QualityRuntime) -> Self { + runtime.pass().started_at = Some(now_ms()); + Self(runtime.clone()) + } +} + +impl Drop for Passing { + fn drop(&mut self) { + self.0.pass().started_at = None; + } +} + +/// Counts one thing for as long as it lives: a file of a library being +/// decoded, or a measurement held back by a scan. +enum Counting { + Measuring(QualityRuntime, Uuid), + Waiting(QualityRuntime), +} + +impl Counting { + fn measuring(runtime: &QualityRuntime, library_id: Uuid) -> Self { + *runtime.0.measuring.entry(library_id).or_insert(0) += 1; + Self::Measuring(runtime.clone(), library_id) + } + + fn waiting(runtime: &QualityRuntime) -> Self { + runtime.0.waiting.fetch_add(1, Ordering::SeqCst); + Self::Waiting(runtime.clone()) + } +} + +impl Drop for Counting { + fn drop(&mut self) { + match self { + Self::Measuring(runtime, library_id) => { + runtime.0.measuring.remove_if_mut(library_id, |_, count| { + *count = count.saturating_sub(1); + *count == 0 + }); + } + Self::Waiting(runtime) => { + runtime.0.waiting.fetch_sub(1, Ordering::SeqCst); + } + } + } +} + /// A file waiting for its measurement. struct Pending { track_id: String, + library_id: Uuid, full_hash: String, root_path: String, relative_path: String, @@ -278,9 +388,6 @@ struct Measured { /// How many files the pass asks for at a time. const PAGE: i64 = 64; -/// How many rows one write carries. -const BATCH: usize = 16; - /// The longest one file may take, whatever it says about itself. The deadline /// is a minute plus the track's duration, and the duration is read from the /// file's own header: without this ceiling, a file claiming ten hours that @@ -352,6 +459,7 @@ impl DomainServices { if parallelism == 0 { return Ok(0); } + let _passing = Passing::begin(&self.quality); let mut measured = 0; // Keyset rather than offset: rows written behind the cursor drop out // of the query, and a file that keeps failing to be written cannot @@ -363,21 +471,22 @@ impl DomainServices { break; }; after = (last.created_at, last.track_id.clone()); - // Each batch is written as soon as it is complete, so a restart - // loses at most one batch of work, and a batch that cannot be - // written costs that batch and not the pass: its files are still - // pending and the next pass asks for them again. - let mut batches = stream::iter(page) + // Each measurement is written as soon as it finishes, so the count + // a reader watches moves file by file and a restart loses at most + // the files in flight. One that cannot be written costs that file + // and not the pass: it is still pending, and the next pass asks + // for it again. + let mut finished = stream::iter(page) .map(|pending| self.measure_one(pending)) - .buffer_unordered(parallelism) - .chunks(BATCH); - while let Some(batch) = batches.next().await { - let batch: Vec = batch.into_iter().flatten().collect(); - let count = batch.len(); - match self.write_measurements(&decoder, batch).await { - Ok(()) => measured += count, + .buffer_unordered(parallelism); + while let Some(done) = finished.next().await { + let Some((done, _measuring)) = done else { + continue; + }; + match self.write_measurements(&decoder, vec![done]).await { + Ok(()) => measured += 1, Err(error) => { - tracing::warn!(error = %error, "a batch of quality measurements was not written"); + tracing::warn!(error = %error, "a quality measurement was not written"); } } } @@ -388,7 +497,8 @@ impl DomainServices { /// One page of the files with nothing current, oldest first. async fn pending_quality(&self, after: &(i64, String)) -> Result, ServiceError> { let rows = sqlx::query(concat!( - "SELECT t.id, t.full_hash, l.root_path, t.relative_path, t.duration_ms, t.created_at \ + "SELECT t.id, t.library_id, t.full_hash, l.root_path, t.relative_path, \ + t.duration_ms, t.created_at \ FROM track t JOIN library l ON l.id = t.library_id", quality_join!(), " WHERE t.is_available = 1 AND ", @@ -403,8 +513,11 @@ impl DomainServices { .await?; rows.into_iter() .map(|row| { + let library_id: String = row.try_get("library_id")?; Ok(Pending { track_id: row.try_get("id")?, + library_id: Uuid::parse_str(&library_id) + .map_err(|error| sqlx::Error::Decode(Box::new(error)))?, full_hash: row.try_get("full_hash")?, root_path: row.try_get("root_path")?, relative_path: row.try_get("relative_path")?, @@ -419,9 +532,20 @@ impl DomainServices { /// Measures one file once no scan is running. `None` when the file cannot /// be reached at all: a drive that is not mounted says nothing about the /// file, so nothing is recorded and the next pass asks again. - async fn measure_one(&self, pending: Pending) -> Option { + /// + /// The file counts as in flight for its library until the caller drops + /// the guard handed back with it, which is once its row is written. + async fn measure_one(&self, pending: Pending) -> Option<(Measured, Counting)> { + // In flight from the moment the pass picks the file, before any wait: + // a scan the caller cannot see must change nothing they are shown, + // and a file that stopped counting while it waited would say a scan + // had started somewhere. + let measuring = Counting::measuring(&self.quality, pending.library_id); let mut scans = self.scanner.scans_in_flight(); - let _ = scans.wait_for(|running| *running == 0).await; + if *scans.borrow_and_update() > 0 { + let _waiting = Counting::waiting(&self.quality); + let _ = scans.wait_for(|running| *running == 0).await; + } let path = crate::media::secure_track_path( std::path::Path::new(&pending.root_path), &pending.relative_path, @@ -433,15 +557,18 @@ impl DomainServices { .min(MAX_DEADLINE); let measurement = quality::measure(&self.ffprobe_path, &self.ffmpeg_path, &path, deadline).await; - Some(Measured { - track_id: pending.track_id, - full_hash: pending.full_hash, - measurement, - }) + Some(( + Measured { + track_id: pending.track_id, + full_hash: pending.full_hash, + measurement, + }, + measuring, + )) } - /// Writes one batch in one transaction, under the writer gate. A row is - /// written only while the track still holds the content it measured. + /// Writes measurements in one transaction, under the writer gate. A row + /// is written only while the track still holds the content it measured. async fn write_measurements( &self, decoder: &quality::Decoder, @@ -534,10 +661,14 @@ impl DomainServices { let not_lossless = count(QualityVerdict::Lossy) + count(QualityVerdict::NotMeasured); let total: i64 = rows.iter().map(|(_, count)| count).sum(); let lossless = total - not_lossless; + let analysis = self + .quality_analysis(user_id, library_id, count(QualityVerdict::Pending)) + .await?; Ok(QualitySummary { enabled: self.quality.0.parallelism > 0, lossless, measured: lossless - count(QualityVerdict::Pending), + analysis, verdicts: QualityVerdict::ALL .into_iter() .map(|verdict| QualityVerdictCount { @@ -722,6 +853,71 @@ impl DomainServices { }) } + /// Where the analysis stands for one library, as its member may read it. + /// + /// A scan holds every measurement back, of every library, but the page + /// says "waiting for the scan" only when the caller can see a library being + /// scanned. A scan of somebody else's library is not something to confirm + /// to them, so it changes nothing they are shown: the pass still reads as + /// running, and its file as in flight. + async fn quality_analysis( + &self, + user_id: Uuid, + library_id: Uuid, + pending: i64, + ) -> Result { + let last_measured_at: Option = sqlx::query_scalar(concat!( + "SELECT MAX(ta.analyzed_at) FROM track t", + quality_join!(), + " WHERE t.library_id = ?" + )) + .bind(library_id.to_string()) + .fetch_one(self.db.pool()) + .await?; + let in_flight = self + .quality + .0 + .measuring + .get(&library_id) + .map_or(0, |count| *count) as i64; + let pass_started_at = self.quality.pass().started_at; + let waiting = self.quality.0.waiting.load(Ordering::SeqCst) > 0; + let state = if self.quality.0.parallelism == 0 { + AnalysisState::Off + } else if pending == 0 && in_flight == 0 { + AnalysisState::Idle + } else if self.scanner.is_scanning(library_id) + || (waiting && self.sees_a_scan(user_id).await?) + { + AnalysisState::WaitingForScan + } else if pass_started_at.is_some() { + // Whatever `waiting` says: a pass held back by a scan the caller + // cannot see reads exactly as one that is not, so the state never + // tells them that somebody else's library is being scanned. + AnalysisState::Running + } else { + AnalysisState::Queued + }; + Ok(QualityAnalysis { + state, + in_flight, + pass_started_at: pass_started_at.filter(|_| state == AnalysisState::Running), + last_measured_at, + }) + } + + /// Whether a library the caller belongs to is being scanned. + async fn sees_a_scan(&self, user_id: Uuid) -> Result { + let libraries: Vec = + sqlx::query_scalar("SELECT library_id FROM library_member WHERE user_id = ?") + .bind(user_id.to_string()) + .fetch_all(self.db.pool()) + .await?; + Ok(libraries.iter().any(|library| { + Uuid::parse_str(library).is_ok_and(|library| self.scanner.is_scanning(library)) + })) + } + /// The caller is a member of the library, in any role. async fn require_member(&self, user_id: Uuid, library_id: Uuid) -> Result<(), ServiceError> { sqlx::query_scalar::<_, i64>( diff --git a/tests/quality.rs b/tests/quality.rs index e728974a..c579e87e 100644 --- a/tests/quality.rs +++ b/tests/quality.rs @@ -23,7 +23,7 @@ use uuid::Uuid; use waveflow_server::authentication::now_ms; use waveflow_server::catalog::LibraryRecord; use waveflow_server::database::{LibraryRole, LibraryVisibility}; -use waveflow_server::services::{AttentionCategory, BrowsePage, QualityVerdict}; +use waveflow_server::services::{AnalysisState, AttentionCategory, BrowsePage, QualityVerdict}; use waveflow_server::AppState; struct Shelf { @@ -482,7 +482,7 @@ async fn a_running_scan_holds_the_analysis_back() { genuine(&shelf.music.join("a.flac"), 3); shelf.scan().await; - let scan = shelf.state.scanner.begin_scan(); + let scan = shelf.state.scanner.begin_scan(shelf.library.id); let services = shelf.state.services.clone(); let pass = tokio::spawn(async move { services.analyze_pending_quality().await.unwrap() }); tokio::time::sleep(std::time::Duration::from_millis(1500)).await; @@ -971,3 +971,223 @@ async fn a_stranger_is_told_nothing_even_at_the_bound() { "access is checked before the bound" ); } + +/// The analysis's state, as the library's owner reads it. +async fn analysis(shelf: &Shelf) -> waveflow_server::services::QualityAnalysis { + shelf + .state + .services + .quality_summary(shelf.owner, shelf.library.id) + .await + .unwrap() + .analysis +} + +/// A minute of noise: long enough that a pass over a few of them can be +/// watched while it runs. +fn minute(path: &Path, seed: u32) { + ffmpeg(&[ + "-f", + "lavfi", + "-i", + &noise(44_100, 60, seed), + "-ac", + "2", + "-sample_fmt", + "s16", + "-c:a", + "flac", + path.to_str().unwrap(), + ]); +} + +#[tokio::test] +async fn the_summary_says_where_the_analysis_stands() { + let shelf = shelf("status").await; + genuine(&shelf.music.join("a.flac"), 3); + shelf.scan().await; + + let before = analysis(&shelf).await; + assert_eq!( + before.state, + AnalysisState::Queued, + "files wait, and no pass is on them" + ); + assert_eq!(before.in_flight, 0); + assert_eq!(before.last_measured_at, None); + + let scan = shelf.state.scanner.begin_scan(shelf.library.id); + assert_eq!( + analysis(&shelf).await.state, + AnalysisState::WaitingForScan, + "its own library is being scanned" + ); + drop(scan); + + shelf.analyze().await; + let after = analysis(&shelf).await; + assert_eq!(after.state, AnalysisState::Idle); + assert_eq!(after.in_flight, 0); + assert_eq!(after.pass_started_at, None); + assert!(after.last_measured_at.is_some()); +} + +/// A pass over three files is watched while it runs: it says it is running, +/// names the file it is decoding, and the count moves one file at a time +/// rather than all at once at the end. +#[tokio::test] +async fn a_running_pass_moves_one_file_at_a_time() { + let shelf = shelf("moving").await; + for (index, seed) in [3, 5, 7].into_iter().enumerate() { + minute(&shelf.music.join(format!("{index}.flac")), seed); + } + shelf.scan().await; + + let services = shelf.state.services.clone(); + let pass = tokio::spawn(async move { services.analyze_pending_quality().await.unwrap() }); + let mut seen_running = false; + let mut seen_partial = false; + for _ in 0..2000 { + if pass.is_finished() { + break; + } + let now = analysis(&shelf).await; + if now.state == AnalysisState::Running && now.in_flight == 1 { + assert!(now.pass_started_at.is_some()); + seen_running = true; + } + let rows = shelf.rows().await; + if (1..3).contains(&rows) && !pass.is_finished() { + seen_partial = true; + } + if seen_running && seen_partial { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + assert!( + seen_running, + "a running pass says so, with its file in flight" + ); + assert!( + seen_partial, + "a measurement is written as it finishes, not at the end of the pass" + ); + assert_eq!(pass.await.unwrap(), 3); + assert_eq!(analysis(&shelf).await.state, AnalysisState::Idle); +} + +/// A scan of somebody else's library holds every measurement back, but it +/// changes nothing its outsiders are shown: the pass reads as running and its +/// file as in flight, exactly as with no scan at all. +#[tokio::test] +async fn a_scan_the_caller_cannot_see_is_not_named() { + let shelf = shelf("mine").await; + genuine(&shelf.music.join("a.flac"), 3); + shelf.scan().await; + let stranger = account(&shelf.state, "stranger").await; + let theirs_root = shelf.config.data_dir.join("theirs"); + std::fs::create_dir_all(&theirs_root).unwrap(); + let theirs = shelf + .state + .db + .create_library( + stranger, + "theirs", + &std::fs::canonicalize(&theirs_root).unwrap(), + LibraryVisibility::Private, + now_ms(), + ) + .await + .unwrap(); + + let scan = shelf.state.scanner.begin_scan(theirs); + let services = shelf.state.services.clone(); + let pass = tokio::spawn(async move { services.analyze_pending_quality().await.unwrap() }); + // Long enough for the pass to have reached the scan and stopped there. + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + assert!(!pass.is_finished(), "the pass waits for the scan"); + let held = analysis(&shelf).await; + assert_eq!( + (held.state, held.in_flight), + (AnalysisState::Running, 1), + "nothing says another library is being scanned" + ); + assert!(held.pass_started_at.is_some()); + + drop(scan); + assert_eq!(pass.await.unwrap(), 1); + assert_eq!(analysis(&shelf).await.state, AnalysisState::Idle); +} + +#[tokio::test] +async fn the_route_carries_the_analysis() { + let shelf = shelf("route-status").await; + genuine(&shelf.music.join("a.flac"), 3); + shelf.scan().await; + let router = waveflow_server::app(&shelf.config, shelf.state.clone()); + let token = shelf.token("route-status").await; + let summary = json_body( + send( + &router, + &token, + get(&format!("/api/v2/libraries/{}/quality", shelf.library.id)), + ) + .await, + ) + .await; + assert_eq!(summary["analysis"]["state"], "queued"); + assert_eq!(summary["analysis"]["in_flight"], 0); + assert!(summary["analysis"]["last_measured_at"].is_null()); +} + +/// The header and the list are one answer: for every verdict, the summary +/// counts exactly the files the list carries under it, including when some +/// files are measured, some wait, and a measured one has gone unavailable. +#[tokio::test] +async fn the_summary_counts_what_the_list_lists() { + let shelf = shelf("agree").await; + let scratch = shelf.config.data_dir.join("scratch"); + std::fs::create_dir_all(&scratch).unwrap(); + genuine(&shelf.music.join("genuine.flac"), 3); + from_mp3(&shelf.music.join("from-mp3.flac"), &scratch); + padded(&shelf.music.join("padded.flac"), &scratch); + genuine(&shelf.music.join("gone.flac"), 9); + std::fs::copy(scratch.join("from-mp3.mp3"), shelf.music.join("lossy.mp3")).unwrap(); + shelf.scan().await; + shelf.analyze().await; + // Measured, then gone: its row stays, and neither side may count it. + std::fs::remove_file(shelf.music.join("gone.flac")).unwrap(); + // Added after the pass: pending on both sides. + genuine(&shelf.music.join("late.flac"), 11); + shelf.scan().await; + + let summary = shelf + .state + .services + .quality_summary(shelf.owner, shelf.library.id) + .await + .unwrap(); + let listed = shelf + .state + .services + .quality_tracks( + shelf.owner, + shelf.library.id, + None, + None, + BrowsePage::default(), + ) + .await + .unwrap(); + for entry in &summary.verdicts { + let in_list = listed + .iter() + .filter(|track| track.quality.verdict == entry.verdict) + .count() as i64; + assert_eq!(entry.count, in_list, "{:?}", entry.verdict); + } + assert_eq!(listed.len(), 5); + assert_eq!(summary.lossless, 4); + assert_eq!(summary.measured, 3); +} diff --git a/webapp/e2e/studio-nocturne.spec.ts b/webapp/e2e/studio-nocturne.spec.ts index ed263b70..3aed9511 100644 --- a/webapp/e2e/studio-nocturne.spec.ts +++ b/webapp/e2e/studio-nocturne.spec.ts @@ -6222,14 +6222,31 @@ async function serveQuality(page: Page) { asked: [] as string[], writes: [] as string[], spectrograms: 0, + summaries: 0, + /** What the summary says, one entry per ask; the last one repeats. */ + analyses: [ + { + measured: 2, + analysis: { + state: "idle", + in_flight: 0, + pass_started_at: null, + last_measured_at: 1_700_000_000_000, + }, + }, + ] as Array<{ measured: number; analysis: Record }>, }; const base = "**/api/v2/libraries/library-1"; - await page.route(`${base}/quality`, (route) => - route.fulfill({ + await page.route(`${base}/quality`, (route) => { + const now = + state.analyses[Math.min(state.summaries, state.analyses.length - 1)]; + state.summaries += 1; + return route.fulfill({ json: { enabled: true, lossless: 3, - measured: 2, + measured: now?.measured ?? 2, + analysis: now?.analysis, verdicts: [ { verdict: "lossy_source", count: 1 }, { verdict: "upsampled", count: 0 }, @@ -6242,8 +6259,8 @@ async function serveQuality(page: Page) { { verdict: "not_measured", count: 0 }, ], }, - }), - ); + }); + }); await page.route(`${base}/quality/tracks?*`, (route) => { const url = new URL(route.request().url()); state.asked.push(url.search); @@ -6317,6 +6334,8 @@ test("lists a library's files by what their audio says, worst first", async ({ const rows = page.locator(".quality-table tbody tr"); await expect(rows).toHaveCount(2); await expect(rows.first()).toContainText("Made From An MP3"); + // The cover before the title: a name alone is hard to place. + await expect(rows.first().locator(".song-cover")).toHaveCount(1); await expect(rows.first()).toContainText("16.4 kHz"); await expect(rows.first()).toContainText("Probably from a lossy file"); await expect(rows.nth(1)).toContainText("Full band"); @@ -6391,3 +6410,128 @@ test("keeps the quality page readable in the light", async ({ page }) => { await expect(page.locator(".quality-table tbody tr")).toHaveCount(2); await expectNoWcagViolations(page); }); + +test("shows the analysis moving, then at rest, and stops asking", async ({ + page, +}) => { + const state = await serveQuality(page); + const running = (measured: number) => ({ + measured, + analysis: { + state: "running", + in_flight: 1, + pass_started_at: 1_700_000_000_000, + last_measured_at: 1_700_000_000_000, + }, + }); + state.analyses = [ + running(1), + running(2), + { + measured: 3, + analysis: { + state: "idle", + in_flight: 0, + pass_started_at: null, + last_measured_at: 1_700_000_000_000, + }, + }, + ]; + await page.goto("/quality"); + const status = page.getByRole("status").filter({ hasText: /Analysing|Idle/ }); + await expect(status).toHaveText( + "Analysing now: 1 of 3 done, 1 in progress.", + ); + // The count moves under the reader while the pass runs. + await expect(status).toHaveText( + "Analysing now: 2 of 3 done, 1 in progress.", + { timeout: 10_000 }, + ); + await expect(status).toHaveText(/^Idle\. Last file analysed /, { + timeout: 10_000, + }); + // At rest, nothing is asked again. + const asked = state.summaries; + await page.waitForTimeout(4_000); + expect(state.summaries).toBe(asked); + await expectNoWcagViolations(page); +}); + +test("says the analysis waits for the scan", async ({ page }) => { + const state = await serveQuality(page); + state.analyses = [ + { + measured: 0, + analysis: { + state: "waiting_for_scan", + in_flight: 0, + pass_started_at: null, + last_measured_at: null, + }, + }, + ]; + await page.goto("/quality"); + await expect( + page.getByText("Waiting for the scan to finish before analysing."), + ).toBeVisible(); +}); + +/** What the summary says of the analysis, at rest or at work. */ +function analysisAt( + measured: number, + state: "idle" | "running", +): { measured: number; analysis: Record } { + return { + measured, + analysis: { + state, + in_flight: state === "running" ? 1 : 0, + pass_started_at: state === "running" ? 1_700_000_000_000 : null, + last_measured_at: 1_700_000_000_000, + }, + }; +} + +test("offers no new analysis while one runs", async ({ page }) => { + const state = await serveQuality(page); + state.analyses = [analysisAt(1, "running")]; + await page.goto("/quality"); + await expect(page.getByText(/^Analysing now:/)).toBeVisible(); + // The server says a pass runs, so there is nothing to start: the offer + // comes from its answer, not from what this page remembers doing. + await expect( + page.getByRole("button", { name: "Analyse again" }), + ).toHaveCount(0); +}); + +test("asks for the header again when the list is filtered", async ({ + page, +}) => { + const state = await serveQuality(page); + state.analyses = [analysisAt(0, "idle"), analysisAt(2, "idle")]; + await page.goto("/quality"); + await expect(page.getByText("0 of 3 lossless files analysed")).toBeVisible(); + await page + .getByRole("list", { name: "Verdicts" }) + .getByRole("link", { name: /No sign of a lossy source/ }) + .click(); + // At rest the summary is not polled, so only the filter can have asked. + await expect(page.getByText("2 of 3 lossless files analysed")).toBeVisible(); +}); + +test("asks for the header again on every visit", async ({ + page, +}, testInfo) => { + test.skip(testInfo.project.name === "mobile", "walks the desktop sidebar"); + const state = await serveQuality(page); + state.analyses = [analysisAt(0, "idle"), analysisAt(2, "idle")]; + await page.goto("/quality"); + await expect(page.getByText("0 of 3 lossless files analysed")).toBeVisible(); + // Away and back inside the cache's thirty seconds: an answer held from the + // first visit would contradict the rows drawn on the second. + const sidebar = page.getByRole("navigation"); + await sidebar.getByRole("link", { name: "Queue" }).click(); + await expect(page.getByRole("heading", { name: "Queue" })).toBeVisible(); + await sidebar.getByRole("link", { name: "Quality" }).click(); + await expect(page.getByText("2 of 3 lossless files analysed")).toBeVisible(); +}); diff --git a/webapp/src/api/index.ts b/webapp/src/api/index.ts index 1e0f0445..26f6ce5b 100644 --- a/webapp/src/api/index.ts +++ b/webapp/src/api/index.ts @@ -145,6 +145,8 @@ export { listPlaylists, } from "./playlists"; export type { + AnalysisState, + QualityAnalysis, QualitySummary, QualityTrack, QualityVerdict, diff --git a/webapp/src/api/quality.ts b/webapp/src/api/quality.ts index d5e95a9f..9fe295a6 100644 --- a/webapp/src/api/quality.ts +++ b/webapp/src/api/quality.ts @@ -29,6 +29,24 @@ export const QUALITY_VERDICTS: QualityVerdict[] = [ "not_measured", ]; +/** What the analyser is doing about one library's files. */ +export type AnalysisState = + | "off" + | "idle" + | "queued" + | "waiting_for_scan" + | "running"; + +export type QualityAnalysis = { + state: AnalysisState; + /** How many of this library's files are being decoded right now. */ + in_flight: number; + /** When the running pass began; `null` unless one is running. */ + pass_started_at: number | null; + /** When a file of this library was last measured. */ + last_measured_at: number | null; +}; + export type QualitySummary = { /** `false` when the operator turned the analysis off. */ enabled: boolean; @@ -37,6 +55,8 @@ export type QualitySummary = { /** Of those, how many have a current measurement. */ measured: number; verdicts: Array<{ verdict: QualityVerdict; count: number }>; + /** Where the analysis stands for this library. */ + analysis: QualityAnalysis; }; /** One file's verdict, and the numbers behind it. */ diff --git a/webapp/src/lib/quality.test.ts b/webapp/src/lib/quality.test.ts index 9f6aac07..1227140a 100644 --- a/webapp/src/lib/quality.test.ts +++ b/webapp/src/lib/quality.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import { + analysisIsWorking, frequencyX, highsStop, SPECTRUM_FLOOR_DB, @@ -55,6 +56,20 @@ describe("where the highs stop", () => { }); }); +describe("whether the page keeps asking", () => { + it("asks while the analysis works", () => { + expect(analysisIsWorking("running")).toBe(true); + expect(analysisIsWorking("queued")).toBe(true); + expect(analysisIsWorking("waiting_for_scan")).toBe(true); + }); + + it("stops once it rests", () => { + expect(analysisIsWorking("idle")).toBe(false); + expect(analysisIsWorking("off")).toBe(false); + expect(analysisIsWorking(undefined)).toBe(false); + }); +}); + describe("a verdict's tone", () => { it("draws an accusation apart from a state", () => { expect(verdictTone("lossy_source")).toBe("finding"); diff --git a/webapp/src/lib/quality.ts b/webapp/src/lib/quality.ts index b02df1e8..01ad9617 100644 --- a/webapp/src/lib/quality.ts +++ b/webapp/src/lib/quality.ts @@ -1,4 +1,4 @@ -import type { QualityVerdict } from "../api"; +import type { AnalysisState, QualityVerdict } from "../api"; /** * The pure half of the quality page and of a track's "Real quality" section @@ -53,6 +53,17 @@ export function highsStop( : null; } +/** + * Whether the analysis is at work on a library's files, so the page keeps + * asking how far it has got. Idle and off are answers that do not change by + * themselves, and asking again would only cost a request. + */ +export function analysisIsWorking(state: AnalysisState | undefined): boolean { + return ( + state === "running" || state === "queued" || state === "waiting_for_scan" + ); +} + /** The level the chart reaches down to, in dB under the loudest band. */ export const SPECTRUM_FLOOR_DB = -120; diff --git a/webapp/src/pages/quality-page.tsx b/webapp/src/pages/quality-page.tsx index 1402fbf0..259f8860 100644 --- a/webapp/src/pages/quality-page.tsx +++ b/webapp/src/pages/quality-page.tsx @@ -8,24 +8,30 @@ import { useEffect, useMemo, useRef, useState } from "react"; import { QUALITY_VERDICTS, + type QualityAnalysis, type QualityTrack, type QualityVerdict, requestQualityAnalysis, } from "../api"; import { qualitySummaryQuery, qualityTracksQuery } from "../api/queries"; +import { Artwork } from "../components/artwork"; import { containerName, kilohertz } from "../components/file-quality"; import { Loading, PageHeader } from "../components/page-shell"; import { VERDICT_LABEL } from "../components/quality-section"; import { TrackProperties } from "../components/track-properties"; -import { highsStop, verdictTone } from "../lib/quality"; +import { analysisIsWorking, highsStop, verdictTone } from "../lib/quality"; import { useI18n } from "../state/i18n"; import { mayCorrectTracks, useLibraryScope } from "../state/library-scope"; /** How long "Analyse again" waits for the second click, as Split does. */ const CONFIRM_MS = 4000; -/** How often the counts are asked again while files wait for analysis. */ -const REFRESH_MS = 10_000; +/** + * How often the counts are asked again while the analysis works. Short + * enough that the count reads as moving: a file takes a fraction of a second + * on a release build, and the reader is watching for exactly that. + */ +const REFRESH_MS = 3_000; /** * The formats a reader filters by, as the scanner names them. ALAC is named @@ -69,14 +75,29 @@ export function QualityPage({ const summary = useQuery({ ...qualitySummaryQuery(libraryId), enabled: active !== null, - refetchInterval: (query) => { - const data = query.state.data; - return data?.enabled && data.measured < data.lossless - ? REFRESH_MS - : false; - }, + // Asked again on every arrival, with the list: the analysis moves in the + // background, and a header held from a visit thirty seconds ago would + // contradict the rows drawn fresh under it. + staleTime: 0, + refetchInterval: (query) => + analysisIsWorking(query.state.data?.analysis?.state) ? REFRESH_MS : false, }); const client = useQueryClient(); + // And the counts follow the list: a filter asks for a list the cache does + // not hold yet, so the header is asked again beside it. Not on the first + // render, which has just asked for both. + const filtered = useRef(false); + useEffect(() => { + void verdict; + void codec; + if (!filtered.current) { + filtered.current = true; + return; + } + void client.invalidateQueries({ + queryKey: ["quality", libraryId, "summary"], + }); + }, [verdict, codec, libraryId, client]); // The list follows the counts: a file measured in the background moves. Only // when the count moves, not on its first answer, which the list has just // been asked alongside. @@ -112,7 +133,10 @@ export function QualityPage({ title={t("nav.quality")} detail={detail} > - {active && mayCorrectTracks(active) ? ( + {active && + mayCorrectTracks(active) && + summary.isFetchedAfterMount && + summary.data?.analysis?.state === "idle" ? ( ) : null} @@ -130,6 +154,13 @@ export function QualityPage({ return (
{header} + {summary.data?.analysis ? ( + + ) : null}

{t("quality.help")}

{!summary.data ? ( @@ -187,6 +218,61 @@ export function QualityPage({ ); } +/** + * Where the analysis stands, in one line: what it is doing, and the count + * moving under it while it works. Announced politely, so a screen reader + * hears the state change without being interrupted on every file. + */ +function AnalysisStatus({ + analysis, + measured, + lossless, +}: { + analysis: QualityAnalysis; + measured: number; + lossless: number; +}) { + const { t, locale } = useI18n(); + const number = new Intl.NumberFormat(locale); + const when = (instant: number) => + new Intl.DateTimeFormat(locale, { + dateStyle: "medium", + timeStyle: "short", + }).format(instant); + const text = (() => { + switch (analysis.state) { + case "running": + return t("quality.status.running", { + measured: number.format(measured), + lossless: number.format(lossless), + inFlight: number.format(analysis.in_flight), + }); + case "waiting_for_scan": + return t("quality.status.waitingForScan"); + case "queued": + return t("quality.status.queued"); + case "off": + return t("quality.disabled"); + default: + return analysis.last_measured_at === null + ? t("quality.status.idleNever") + : t("quality.status.idle", { when: when(analysis.last_measured_at) }); + } + })(); + return ( +

+ {analysisIsWorking(analysis.state) ? ( +

+ ); +} + function QualityList({ libraryId, verdict, @@ -197,9 +283,11 @@ function QualityList({ codec: string | undefined; }) { const { t, locale } = useI18n(); - const tracks = useInfiniteQuery( - qualityTracksQuery(libraryId, { verdict, codec }), - ); + const tracks = useInfiniteQuery({ + ...qualityTracksQuery(libraryId, { verdict, codec }), + // Asked again on every arrival, as the header is: see `QualityPage`. + staleTime: 0, + }); const songs = useMemo(() => tracks.data?.pages.flat() ?? [], [tracks.data]); const [opened, setOpened] = useState(null); const returnTo = useRef(null); @@ -249,6 +337,11 @@ function QualityList({ setOpened(song); }} > + {song.title} diff --git a/webapp/src/state/i18n.tsx b/webapp/src/state/i18n.tsx index 8fbc7fc9..ac083b6c 100644 --- a/webapp/src/state/i18n.tsx +++ b/webapp/src/state/i18n.tsx @@ -496,6 +496,14 @@ const en = { "quality.progress": "{measured} of {lossless} lossless files analysed", "quality.disabled": "The analysis is turned off on this server.", "quality.noLossless": "No lossless file in this library.", + "quality.status.running": + "Analysing now: {measured} of {lossless} done, {inFlight} in progress.", + "quality.status.waitingForScan": + "Waiting for the scan to finish before analysing.", + "quality.status.queued": + "Queued: the analysis will reach these files shortly.", + "quality.status.idle": "Idle. Last file analysed {when}.", + "quality.status.idleNever": "Idle. No file has been analysed yet.", "quality.filters": "Verdicts", "quality.all": "All files", "quality.format": "Format", @@ -1323,6 +1331,14 @@ const fr: Record = { "quality.progress": "{measured} fichiers sans perte analysés sur {lossless}", "quality.disabled": "L’analyse est désactivée sur ce serveur.", "quality.noLossless": "Aucun fichier sans perte dans cette bibliothèque.", + "quality.status.running": + "Analyse en cours : {measured} sur {lossless} terminés, {inFlight} en traitement.", + "quality.status.waitingForScan": + "En attente de la fin du scan avant d’analyser.", + "quality.status.queued": + "En file d’attente : l’analyse arrivera bientôt à ces fichiers.", + "quality.status.idle": "Au repos. Dernier fichier analysé le {when}.", + "quality.status.idleNever": "Au repos. Aucun fichier n’a encore été analysé.", "quality.filters": "Verdicts", "quality.all": "Tous les fichiers", "quality.format": "Format", diff --git a/webapp/src/styles/styles.css b/webapp/src/styles/styles.css index 7e5d0eb0..9d88c91e 100644 --- a/webapp/src/styles/styles.css +++ b/webapp/src/styles/styles.css @@ -3471,6 +3471,47 @@ button.danger { list-style: none; } +/* Where the analysis stands. The dot pulses while it works, and is still for + whoever asked the system for less motion: the words say the same thing. */ +.quality-status { + display: flex; + align-items: center; + gap: var(--space-2); + margin: 0 0 var(--space-3); + font-size: var(--text-sm); +} + +.quality-status.idle, +.quality-status.off { + color: var(--muted); +} + +.quality-status-dot { + flex: none; + width: 0.55rem; + height: 0.55rem; + border-radius: 50%; + background: var(--accent); + animation: quality-pulse 1.4s ease-in-out infinite; +} + +.quality-status.waiting_for_scan .quality-status-dot, +.quality-status.queued .quality-status-dot { + background: var(--muted); +} + +@keyframes quality-pulse { + 50% { + opacity: 0.35; + } +} + +@media (prefers-reduced-motion: reduce) { + .quality-status-dot { + animation: none; + } +} + .quality-format { display: inline-flex; align-items: center;