diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f9a18a6..cf47d6ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -413,6 +413,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 +### Analytics โ€” advanced library backfill without webview jank + +**By [@cucadmuh](https://github.com/cucadmuh), PR [#881](https://github.com/Psychotoxical/psysonic/pull/881)** + +* **Settings โ†’ Library โ†’ Analytics โ†’ Advanced** on large libraries no longer stalls the whole UI (~4 FPS): catalog scheduling runs in a native background worker like cover backfill, not a webview polling loop. +* Partially analyzed tracks (hash + BPM but missing waveform/loudness) are picked up via a targeted second scan with a reset cursor so the library is not skipped mid-pass. +* Performance Probe analysis stats still update during background backfill; waveform/enrichment refresh events stay quiet for low-priority work. + + + ## [1.46.0] - 2026-05-18 > **๐Ÿ™ Special thanks to [@zz5zz](https://github.com/zz5zz)** for his tireless quirk-spotting and bug reports on the [Psysonic Discord](https://discord.gg/AMnDRErm4u) โ€” several of the polish fixes in this release landed directly off the back of his messages. diff --git a/src-tauri/crates/psysonic-analysis/src/analysis_cache/compute.rs b/src-tauri/crates/psysonic-analysis/src/analysis_cache/compute.rs index bc88f75b..c16b4943 100644 --- a/src-tauri/crates/psysonic-analysis/src/analysis_cache/compute.rs +++ b/src-tauri/crates/psysonic-analysis/src/analysis_cache/compute.rs @@ -47,6 +47,7 @@ pub fn seed_from_bytes_execute( server_id: &str, track_id: &str, bytes: &[u8], + notify_ui: bool, ) -> Result<(SeedFromBytesOutcome, AnalysisSeedTimings), String> { let seed_started = Instant::now(); let Some(cache) = app.try_state::() else { @@ -84,6 +85,7 @@ pub fn seed_from_bytes_execute( server_id, track_id, bytes, + notify_ui, ); if matches!(enrichment_outcome, TrackEnrichmentOutcome::Failed) { let key = TrackKey { @@ -1233,7 +1235,7 @@ mod tests { let app = tauri::test::mock_app(); let wav = build_mono_pcm16_wav(&sine_440_at_minus_6db(44_100, 0.25), 44_100); let handle = app.handle().clone(); - let (outcome, timings) = seed_from_bytes_execute(&handle, "s", "t", &wav) + let (outcome, timings) = seed_from_bytes_execute(&handle, "s", "t", &wav, true) .expect("seed execute should return a graceful skip"); assert_eq!(outcome, SeedFromBytesOutcome::SkippedNoAnalysisCache); assert_eq!(timings.seed_ms, 0); @@ -1248,12 +1250,12 @@ mod tests { let handle = app.handle().clone(); let (first, timings_first) = - seed_from_bytes_execute(&handle, "server-a", "track-exec", &wav).unwrap(); + seed_from_bytes_execute(&handle, "server-a", "track-exec", &wav, true).unwrap(); assert_eq!(first, SeedFromBytesOutcome::Upserted); assert!(timings_first.seed_ms <= 30_000); let (second, timings_second) = - seed_from_bytes_execute(&handle, "server-a", "track-exec", &wav).unwrap(); + seed_from_bytes_execute(&handle, "server-a", "track-exec", &wav, true).unwrap(); assert_eq!(second, SeedFromBytesOutcome::SkippedWaveformCacheHit); assert!(timings_second.seed_ms <= 30_000); } diff --git a/src-tauri/crates/psysonic-analysis/src/analysis_runtime.rs b/src-tauri/crates/psysonic-analysis/src/analysis_runtime.rs index 164817b0..73f85cbc 100644 --- a/src-tauri/crates/psysonic-analysis/src/analysis_runtime.rs +++ b/src-tauri/crates/psysonic-analysis/src/analysis_runtime.rs @@ -4,6 +4,7 @@ use std::sync::{Arc, Mutex, OnceLock}; use tauri::{Emitter, Manager}; +use psysonic_core::ports::PlaybackQueryHandle; use psysonic_core::user_agent::subsonic_wire_user_agent; use psysonic_core::track_enrichment::TrackEnrichmentOutcome; @@ -436,7 +437,14 @@ async fn enqueue_track_analysis_with_fetch( content_hash ); let bpm_started = std::time::Instant::now(); - let outcome = run_track_enrichment_from_bytes(app, server_id, track_id, bytes).await; + let outcome = run_track_enrichment_from_bytes( + app, + server_id, + track_id, + bytes, + analysis_emits_ui_events(priority), + ) + .await; if matches!(outcome, TrackEnrichmentOutcome::Failed) { if let Some(cache) = app.try_state::() { let key = analysis_cache::TrackKey { @@ -464,6 +472,7 @@ pub async fn run_track_enrichment_from_bytes( server_id: &str, track_id: &str, bytes: &[u8], + notify_ui: bool, ) -> TrackEnrichmentOutcome { if server_id.is_empty() { return TrackEnrichmentOutcome::SkippedNoServer; @@ -473,7 +482,7 @@ pub async fn run_track_enrichment_from_bytes( let tid = track_id.to_string(); let data = bytes.to_vec(); match tokio::task::spawn_blocking(move || { - crate::track_enrichment::run_track_enrichment_if_needed(&app, &sid, &tid, &data) + crate::track_enrichment::run_track_enrichment_if_needed(&app, &sid, &tid, &data, notify_ui) }) .await { @@ -795,6 +804,118 @@ pub fn analysis_backfill_resolve_priority( AnalysisBackfillPriority::Low } +/// Library backfill uses `Low` โ€” skip waveform / enrichment refresh IPC (`analysis:track-perf` still emits for probes). +pub fn analysis_emits_ui_events(priority: AnalysisBackfillPriority) -> bool { + !matches!(priority, AnalysisBackfillPriority::Low) +} + +/// Enqueue HTTP download + analysis seed (native coordinator + optional UI invoke). +pub fn enqueue_seed_from_url( + app: &tauri::AppHandle, + track_id: &str, + url: &str, + server_id_hint: Option<&str>, + explicit_priority: Option, + force: bool, +) -> Result<(), String> { + if track_id.trim().is_empty() || url.trim().is_empty() { + return Ok(()); + } + let server_id = if let Ok(parsed) = reqwest::Url::parse(url) { + if parsed.scheme() == "http" || parsed.scheme() == "https" { + let host = parsed.host_str().unwrap_or_default(); + let mut base_path = parsed.path().to_string(); + if let Some(idx) = base_path.find("/rest") { + base_path.truncate(idx); + } + while base_path.ends_with('/') { + base_path.pop(); + } + if host.is_empty() { + server_id_hint.unwrap_or("").to_string() + } else { + let mut base = host.to_string(); + if let Some(port) = parsed.port() { + base.push_str(&format!(":{port}")); + } + if !base_path.is_empty() { + base.push_str(&base_path); + } + base + } + } else { + server_id_hint.unwrap_or("").to_string() + } + } else { + server_id_hint.unwrap_or("").to_string() + }; + if !force { + if let Some(playback) = app.try_state::() { + if playback.ranged_loudness_backfill_should_defer(track_id) { + crate::app_deprintln!( + "[analysis] backfill skip track_id={} reason=ranged_playback_will_seed", + track_id + ); + return Ok(()); + } + } + } + if !force { + if let Some(cache) = app.try_state::() { + if cache.cpu_seed_redundant_for_track(&server_id, track_id)? { + if server_id.is_empty() { + crate::app_deprintln!( + "[analysis] backfill skip (no server scope): {}", + track_id + ); + return Ok(()); + } + if !track_analysis_needs_work(app, &server_id, track_id)? { + crate::app_deprintln!( + "[analysis] backfill skip (analysis complete): {}", + track_id + ); + return Ok(()); + } + crate::app_deprintln!( + "[analysis] backfill enqueue (analysis pending) track_id={}", + track_id + ); + } + } + } + let tid_log = track_id.to_string(); + let resolved = analysis_backfill_resolve_priority(app, &server_id, track_id, explicit_priority); + let shared = analysis_backfill_shared(app); + let kind = { + let mut st = shared + .state + .lock() + .map_err(|_| "analysis backfill lock poisoned".to_string())?; + st.enqueue(server_id, track_id.to_string(), url.to_string(), resolved) + }; + match kind { + AnalysisBackfillEnqueueKind::NewLow + | AnalysisBackfillEnqueueKind::NewMiddle + | AnalysisBackfillEnqueueKind::NewHigh => { + shared.ping_worker(); + crate::app_deprintln!( + "[analysis] backfill enqueued: track_id={} priority={resolved:?}", + tid_log, + ); + } + AnalysisBackfillEnqueueKind::ReorderedHigher => { + shared.ping_worker(); + crate::app_deprintln!( + "[analysis] backfill bumped tier track_id={} priority={resolved:?}", + tid_log, + ); + } + AnalysisBackfillEnqueueKind::DuplicateSkipped | AnalysisBackfillEnqueueKind::RunningSkipped => {} + } + Ok(()) +} + // โ”€โ”€โ”€ Full-track waveform + loudness: CPU seed queue (parallel decode workers) โ”€ #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -1131,12 +1252,19 @@ async fn spawn_cpu_seed_slots(app: &tauri::AppHandle, shared: &Arc, app: tauri::AppHandle, ) -> Result<(), String> { - if track_id.trim().is_empty() || url.trim().is_empty() { - return Ok(()); - } - let server_id = if let Ok(parsed) = reqwest::Url::parse(&url) { - if parsed.scheme() == "http" || parsed.scheme() == "https" { - let host = parsed.host_str().unwrap_or_default(); - let mut base_path = parsed.path().to_string(); - if let Some(idx) = base_path.find("/rest") { - base_path.truncate(idx); - } - while base_path.ends_with('/') { - base_path.pop(); - } - if host.is_empty() { - server_id.unwrap_or_default() - } else { - let mut base = host.to_string(); - if let Some(port) = parsed.port() { - base.push_str(&format!(":{port}")); - } - if !base_path.is_empty() { - base.push_str(&base_path); - } - base - } - } else { - server_id.unwrap_or_default() - } - } else { - server_id.unwrap_or_default() - }; - let force = force.unwrap_or(false); - if !force { - if let Some(playback) = app.try_state::() { - if playback.ranged_loudness_backfill_should_defer(&track_id) { - crate::app_deprintln!( - "[analysis] backfill skip track_id={} reason=ranged_playback_will_seed", - track_id - ); - return Ok(()); - } - } - } - if !force { - if let Some(cache) = app.try_state::() { - if cache.cpu_seed_redundant_for_track(&server_id, &track_id)? { - if server_id.is_empty() { - crate::app_deprintln!( - "[analysis] backfill skip (no server scope): {}", - track_id - ); - return Ok(()); - } - if !track_analysis_needs_work(&app, &server_id, &track_id)? { - crate::app_deprintln!( - "[analysis] backfill skip (analysis complete): {}", - track_id - ); - return Ok(()); - } - crate::app_deprintln!( - "[analysis] backfill enqueue (analysis pending) track_id={}", - track_id - ); - } - } - } - let tid_log = track_id.clone(); let explicit = AnalysisBackfillPriority::from_optional_str(priority.as_deref()); - let resolved = - analysis_backfill_resolve_priority(&app, &server_id, &track_id, explicit); - let shared = analysis_backfill_shared(&app); - let kind = { - let mut st = shared - .state - .lock() - .map_err(|_| "analysis backfill lock poisoned".to_string())?; - st.enqueue(server_id, track_id, url, resolved) - }; - match kind { - AnalysisBackfillEnqueueKind::NewLow - | AnalysisBackfillEnqueueKind::NewMiddle - | AnalysisBackfillEnqueueKind::NewHigh => { - shared.ping_worker(); - crate::app_deprintln!( - "[analysis] backfill enqueued: track_id={} priority={resolved:?}", - tid_log, - ); - } - AnalysisBackfillEnqueueKind::ReorderedHigher => { - shared.ping_worker(); - crate::app_deprintln!( - "[analysis] backfill bumped tier track_id={} priority={resolved:?}", - tid_log, - ); - } - AnalysisBackfillEnqueueKind::DuplicateSkipped | AnalysisBackfillEnqueueKind::RunningSkipped => {} - } - Ok(()) + enqueue_seed_from_url( + &app, + &track_id, + &url, + server_id.as_deref(), + explicit, + force.unwrap_or(false), + ) } #[derive(Debug, Clone, serde::Deserialize)] diff --git a/src-tauri/crates/psysonic-analysis/src/track_enrichment.rs b/src-tauri/crates/psysonic-analysis/src/track_enrichment.rs index 54557ba3..83a28f38 100644 --- a/src-tauri/crates/psysonic-analysis/src/track_enrichment.rs +++ b/src-tauri/crates/psysonic-analysis/src/track_enrichment.rs @@ -35,6 +35,7 @@ pub fn run_track_enrichment_if_needed( server_id: &str, track_id: &str, bytes: &[u8], + notify_ui: bool, ) -> TrackEnrichmentOutcome { if server_id.is_empty() { return TrackEnrichmentOutcome::SkippedNoServer; @@ -56,7 +57,9 @@ pub fn run_track_enrichment_if_needed( server_id, content_hash ); - emit_enrichment_updated(app, server_id, track_id); + if notify_ui { + emit_enrichment_updated(app, server_id, track_id); + } TrackEnrichmentOutcome::Applied } Err(e) => { diff --git a/src-tauri/crates/psysonic-integration/src/subsonic/mod.rs b/src-tauri/crates/psysonic-integration/src/subsonic/mod.rs index 36afd40c..ee80c203 100644 --- a/src-tauri/crates/psysonic-integration/src/subsonic/mod.rs +++ b/src-tauri/crates/psysonic-integration/src/subsonic/mod.rs @@ -5,12 +5,14 @@ pub mod auth; pub mod client; pub mod error; +pub mod stream_url; pub mod types; pub use auth::SubsonicCredentials; pub use client::{ fingerprint_sample, SubsonicClient, SUBSONIC_API_VERSION, SUBSONIC_CLIENT_ID, }; +pub use stream_url::{build_stream_view_url, rest_base_from_url}; pub use error::SubsonicError; pub use types::{ Album, AlbumSummary, ArtistIndex, ArtistRef, IndexBucket, ScanStatus, SearchResult, ServerInfo, diff --git a/src-tauri/crates/psysonic-integration/src/subsonic/stream_url.rs b/src-tauri/crates/psysonic-integration/src/subsonic/stream_url.rs new file mode 100644 index 00000000..dacb7d55 --- /dev/null +++ b/src-tauri/crates/psysonic-integration/src/subsonic/stream_url.rs @@ -0,0 +1,61 @@ +//! Subsonic `stream.view` URLs for native library analysis backfill. + +use url::Url; + +use super::auth::SubsonicCredentials; +use super::client::{SUBSONIC_API_VERSION, SUBSONIC_CLIENT_ID}; + +/// `{origin}/rest` โ€” mirrors frontend `restBaseFromUrl`. +pub fn rest_base_from_url(server_url: &str) -> String { + let trimmed = server_url.trim().trim_end_matches('/'); + let base = if trimmed.starts_with("http://") || trimmed.starts_with("https://") { + trimmed.to_string() + } else { + format!("http://{trimmed}") + }; + format!("{base}/rest") +} + +/// Authenticated `stream.view` URL for a library track id. +pub fn build_stream_view_url( + server_url: &str, + username: &str, + password: &str, + track_id: &str, +) -> String { + let creds = SubsonicCredentials::from_password(username, password); + let base = rest_base_from_url(server_url); + let mut url = + Url::parse(&format!("{base}/stream.view")).unwrap_or_else(|_| Url::parse("http://invalid/rest/stream.view").unwrap()); + { + let mut q = url.query_pairs_mut(); + q.append_pair("id", track_id); + q.append_pair("u", &creds.username); + q.append_pair("t", &creds.token); + q.append_pair("s", &creds.salt); + q.append_pair("v", SUBSONIC_API_VERSION); + q.append_pair("c", SUBSONIC_CLIENT_ID); + q.append_pair("f", "json"); + } + url.to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stream_url_contains_track_and_auth_params() { + let url = build_stream_view_url( + "https://music.example", + "alice", + "secret", + "tr-42", + ); + assert!(url.contains("stream.view")); + assert!(url.contains("id=tr-42")); + assert!(url.contains("u=alice")); + assert!(url.contains("&t=")); + assert!(url.contains("&s=")); + } +} diff --git a/src-tauri/crates/psysonic-library/src/analysis_backfill.rs b/src-tauri/crates/psysonic-library/src/analysis_backfill.rs index 1ddbe623..c1dd768b 100644 --- a/src-tauri/crates/psysonic-library/src/analysis_backfill.rs +++ b/src-tauri/crates/psysonic-library/src/analysis_backfill.rs @@ -31,14 +31,12 @@ pub struct LibraryAnalysisProgressDto { enum ScanMode { Candidates, - Full, + /// Tracks with hash + BPM that may still need waveform/LUFS/enrichment gaps. + HashBpmGaps, } -/// Candidate SQL skips tracks that already have `content_hash` and an analysis BPM -/// fact. Those rows can still need waveform/LUFS โ€” the full-table pass must start -/// from the first id, not from the last candidate cursor. -fn begin_full_library_scan() -> (ScanMode, Option) { - (ScanMode::Full, None) +fn begin_hash_bpm_gap_scan() -> (ScanMode, Option) { + (ScanMode::HashBpmGaps, None) } pub fn collect_analysis_backfill_batch( @@ -64,16 +62,18 @@ pub fn collect_analysis_backfill_batch( ScanMode::Candidates => { repo.list_analysis_candidate_ids_after(server_id, after.as_deref(), SCAN_CHUNK)? } - ScanMode::Full => repo.list_track_ids_after(server_id, after.as_deref(), SCAN_CHUNK)?, + ScanMode::HashBpmGaps => { + repo.list_analysis_hash_bpm_ids_after(server_id, after.as_deref(), SCAN_CHUNK)? + } }; if page.is_empty() { match mode { ScanMode::Candidates => { - (mode, after) = begin_full_library_scan(); + (mode, after) = begin_hash_bpm_gap_scan(); continue; } - ScanMode::Full => { + ScanMode::HashBpmGaps => { return Ok(LibraryAnalysisBackfillBatchDto { track_ids: found, next_cursor: after, @@ -105,9 +105,9 @@ pub fn collect_analysis_backfill_batch( if page_len < SCAN_CHUNK { match mode { ScanMode::Candidates => { - (mode, after) = begin_full_library_scan(); + (mode, after) = begin_hash_bpm_gap_scan(); } - ScanMode::Full => { + ScanMode::HashBpmGaps => { return Ok(LibraryAnalysisBackfillBatchDto { track_ids: found, next_cursor: after, diff --git a/src-tauri/crates/psysonic-library/src/analysis_backfill_policy.rs b/src-tauri/crates/psysonic-library/src/analysis_backfill_policy.rs new file mode 100644 index 00000000..796d59f6 --- /dev/null +++ b/src-tauri/crates/psysonic-library/src/analysis_backfill_policy.rs @@ -0,0 +1,64 @@ +//! Library analysis backfill queue watermark โ€” native coordinator (spec: Settings โ†’ Library). + +pub const LIBRARY_BACKLOG_DEPTH_MULTIPLIER: u32 = 3; +pub const LIBRARY_BACKLOG_MIN: u32 = 8; +pub const LIBRARY_BACKLOG_MAX: u32 = 240; +pub const LIBRARY_ANALYSIS_BACKFILL_BATCH_SIZE: u32 = 20; + +#[derive(Debug, Clone, Copy, Default)] +pub struct PipelineBacklogCounts { + pub http_queued: u32, + pub http_download_active: u32, + pub cpu_queued: u32, + pub cpu_decode_active: u32, +} + +impl PipelineBacklogCounts { + pub fn total(self) -> u32 { + self.http_queued + .saturating_add(self.http_download_active) + .saturating_add(self.cpu_queued) + .saturating_add(self.cpu_decode_active) + } +} + +pub fn compute_library_backfill_target_depth(workers: u32) -> u32 { + let w = workers.max(1); + (w * LIBRARY_BACKLOG_DEPTH_MULTIPLIER).clamp(LIBRARY_BACKLOG_MIN, LIBRARY_BACKLOG_MAX) +} + +pub fn library_backfill_needs_top_up(counts: PipelineBacklogCounts, workers: u32) -> bool { + counts.total() < compute_library_backfill_target_depth(workers) +} + +pub fn library_backfill_top_up_limit(counts: PipelineBacklogCounts, workers: u32) -> u32 { + let target = compute_library_backfill_target_depth(workers); + let deficit = target.saturating_sub(counts.total()); + if deficit == 0 { + return 0; + } + deficit.min(LIBRARY_ANALYSIS_BACKFILL_BATCH_SIZE) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn target_depth_uses_floor_and_cap() { + assert_eq!(compute_library_backfill_target_depth(1), 8); + assert_eq!(compute_library_backfill_target_depth(4), 12); + assert_eq!(compute_library_backfill_target_depth(100), 240); + } + + #[test] + fn top_up_when_backlog_below_target() { + let counts = PipelineBacklogCounts { + http_queued: 2, + http_download_active: 1, + ..Default::default() + }; + assert!(library_backfill_needs_top_up(counts, 8)); + assert_eq!(library_backfill_top_up_limit(counts, 8), 20); + } +} diff --git a/src-tauri/crates/psysonic-library/src/lib.rs b/src-tauri/crates/psysonic-library/src/lib.rs index d5a2c070..1279abef 100644 --- a/src-tauri/crates/psysonic-library/src/lib.rs +++ b/src-tauri/crates/psysonic-library/src/lib.rs @@ -13,6 +13,8 @@ pub mod album_compilation_filter; pub mod browse_support; mod advanced_search_mood; pub mod analysis_backfill; +pub mod analysis_backfill_policy; +pub mod library_readiness; pub mod artist_lossless_browse; pub mod cover_backfill; pub mod cover_resolve; diff --git a/src-tauri/crates/psysonic-library/src/library_readiness.rs b/src-tauri/crates/psysonic-library/src/library_readiness.rs new file mode 100644 index 00000000..408bd7b0 --- /dev/null +++ b/src-tauri/crates/psysonic-library/src/library_readiness.rs @@ -0,0 +1,64 @@ +//! When the local library index is safe for background analysis backfill. + +use crate::dto::local_tracks_max_updated_ms; +use crate::repos::{sync_state::SyncStateRepository, TrackRepository}; +use crate::store::LibraryStore; + +/// Mirrors frontend `libraryStatusIsReady` / `libraryIsReady` (spec ยง9.3). +pub fn library_server_is_ready(store: &LibraryStore, server_id: &str) -> Result { + let repo = SyncStateRepository::new(store); + let phase = repo + .get_sync_phase(server_id, "") + .map_err(|e| e.to_string())?; + + let Some(phase) = phase else { + return Ok(has_any_local_tracks(store, server_id)); + }; + + if phase == "ready" { + return Ok(true); + } + if phase == "initial_sync" { + let local = repo + .get_local_track_count(server_id, "") + .map_err(|e| e.to_string())? + .unwrap_or(0); + let server = repo + .get_server_track_count(server_id, "") + .map_err(|e| e.to_string())? + .unwrap_or(0); + if server > 0 && (local as f64 / server as f64) >= 0.95 { + return Ok(true); + } + return Ok(false); + } + if phase == "idle" { + if has_any_local_tracks(store, server_id) { + return Ok(true); + } + if repo + .has_last_full_sync_at(server_id, "") + .map_err(|e| e.to_string())? + { + return Ok(true); + } + if local_tracks_max_updated_ms(store, server_id)? + .is_some_and(|ms| ms > 0) + { + return Ok(true); + } + let local = repo + .get_local_track_count(server_id, "") + .map_err(|e| e.to_string())? + .unwrap_or(0); + return Ok(local > 0); + } + Ok(false) +} + +fn has_any_local_tracks(store: &LibraryStore, server_id: &str) -> bool { + TrackRepository::new(store) + .count_live_tracks(server_id) + .map(|n| n > 0) + .unwrap_or(false) +} diff --git a/src-tauri/crates/psysonic-library/src/repos/track.rs b/src-tauri/crates/psysonic-library/src/repos/track.rs index 30024972..283877c2 100644 --- a/src-tauri/crates/psysonic-library/src/repos/track.rs +++ b/src-tauri/crates/psysonic-library/src/repos/track.rs @@ -292,6 +292,37 @@ impl<'a> TrackRepository<'a> { }) } + /// Tracks with `content_hash` and an analysis BPM fact โ€” may still lack waveform/LUFS. + /// Confirmed per id via [`TrackAnalysisNeedsWorkQuery`]. + pub fn list_analysis_hash_bpm_ids_after( + &self, + server_id: &str, + after_id: Option<&str>, + limit: usize, + ) -> Result, String> { + if limit == 0 { + return Ok(vec![]); + } + let limit = i64::try_from(limit).map_err(|e| e.to_string())?; + self.store.with_read_conn(|conn| { + let sql = "SELECT t.id FROM track t \ + WHERE t.server_id = ?1 AND t.deleted = 0 \ + AND (?2 IS NULL OR t.id > ?2) \ + AND t.content_hash IS NOT NULL \ + AND EXISTS ( \ + SELECT 1 FROM track_fact f \ + WHERE f.server_id = t.server_id \ + AND f.track_id = t.id \ + AND f.fact_kind = 'bpm' \ + AND f.source_kind = 'analysis' \ + ) \ + ORDER BY t.id ASC LIMIT ?3"; + let mut stmt = conn.prepare(sql)?; + let rows = stmt.query_map(params![server_id, after_id, limit], |row| row.get(0))?; + rows.collect::>>() + }) + } + /// Cheap SQL prefilter: tracks that never received a playback hash and/or /// lack an oximedia BPM fact. Full analysis gaps are confirmed per id via /// [`TrackAnalysisNeedsWorkQuery`] in the shell crate. diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index ef024914..ae1c0637 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3,6 +3,7 @@ pub mod cli; mod cover_cache; +mod library_analysis_backfill; mod lib_commands; pub use psysonic_integration::discord; @@ -123,6 +124,9 @@ pub fn run() { cover_cache::init_cover_cache(app.handle()) .map_err(|e| format!("cover cache init failed: {e}"))?; + library_analysis_backfill::init_library_analysis_backfill(app.handle()) + .map_err(|e| format!("library analysis backfill init failed: {e}"))?; + // โ”€โ”€ Library track store (psysonic-library, PR-5a + PR-5b) โ”€โ”€โ”€โ”€โ”€ // PR-5a brought up the read-only Tauri surface + LibraryRuntime. // PR-5b adds the mutating commands, sync session map, current-job @@ -734,6 +738,7 @@ pub fn run() { psysonic_library::commands::library_migrate_server_index_keys, psysonic_library::commands::library_delete_server_data, psysonic_library::commands::library_analysis_backfill_batch, + library_analysis_backfill::library_analysis_backfill_configure, psysonic_library::commands::library_resolve_cover_entry, cover_cache::cover_cache_peek_batch, cover_cache::cover_cache_ensure, diff --git a/src-tauri/src/library_analysis_backfill/mod.rs b/src-tauri/src/library_analysis_backfill/mod.rs new file mode 100644 index 00000000..0d899326 --- /dev/null +++ b/src-tauri/src/library_analysis_backfill/mod.rs @@ -0,0 +1,59 @@ +//! Library analysis backfill โ€” native coordinator (advanced analytics strategy). + +mod worker; + +use std::sync::Arc; + +use tauri::{AppHandle, Manager}; +use worker::{ + spawn_coordinator, setup_library_sync_idle_listener, LibraryAnalysisBackfillSession, + LibraryAnalysisBackfillWorker, +}; + + +pub fn init_library_analysis_backfill(app: &AppHandle) -> Result<(), String> { + let worker = Arc::new(LibraryAnalysisBackfillWorker::new()); + app.manage(worker.clone()); + setup_library_sync_idle_listener(app); + spawn_coordinator(app, worker); + Ok(()) +} + +#[tauri::command] +#[allow(clippy::too_many_arguments)] // Tauri command surface โ€” args map 1:1 to the JS call (like cover configure + workers). +pub async fn library_analysis_backfill_configure( + app: AppHandle, + enabled: bool, + server_index_key: String, + library_server_id: String, + server_url: String, + username: String, + password: String, + workers: u32, +) -> Result<(), String> { + let worker = app + .try_state::>() + .ok_or_else(|| "library analysis backfill worker not initialized".to_string())?; + + let session = if enabled + && !server_index_key.is_empty() + && !library_server_id.is_empty() + && !server_url.is_empty() + { + Some(LibraryAnalysisBackfillSession { + server_index_key, + library_server_id, + server_url, + username, + password, + workers: workers.max(1), + }) + } else { + None + }; + + worker + .set_session(enabled && session.is_some(), session) + .await; + Ok(()) +} diff --git a/src-tauri/src/library_analysis_backfill/worker.rs b/src-tauri/src/library_analysis_backfill/worker.rs new file mode 100644 index 00000000..1b68f162 --- /dev/null +++ b/src-tauri/src/library_analysis_backfill/worker.rs @@ -0,0 +1,343 @@ +//! Native library analysis backfill coordinator (advanced strategy). +//! +//! Replaces the webview `while` loop: top-up HTTP/CPU seed backlog without +//! blocking the UI on `library_analysis_backfill_batch` IPC. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use psysonic_analysis::analysis_runtime::{ + analysis_pipeline_queue_stats, analysis_set_pipeline_parallelism, enqueue_seed_from_url, + AnalysisBackfillPriority, +}; +use psysonic_integration::subsonic::build_stream_view_url; +use psysonic_library::analysis_backfill::{ + collect_analysis_backfill_batch, collect_analysis_progress, LibraryAnalysisBackfillBatchDto, +}; +use psysonic_library::analysis_backfill_policy::{ + library_backfill_needs_top_up, library_backfill_top_up_limit, PipelineBacklogCounts, +}; +use psysonic_library::library_readiness::library_server_is_ready; +use psysonic_library::payload::LibrarySyncProgressPayload; +use psysonic_library::repos::TrackRepository; +use psysonic_library::LibraryRuntime; +use serde::Deserialize; +use tauri::{AppHandle, Listener, Manager}; +use tokio::sync::Mutex; + +const TOP_UP_POLL_MS: u64 = 500; +const STEADY_POLL_MS: u64 = 2000; +const READY_POLL_MS: u64 = 5000; +const EXHAUSTED_PAUSE_MS: u64 = 15_000; +const EXHAUSTED_PENDING_RESCAN_MS: u64 = 2000; +const COMPLETED_RECHECK_MS: u64 = 300_000; +const EXHAUSTED_DONE_STREAK: u32 = 2; +const SYNC_WAIT_MS: u64 = 5000; + +#[derive(Clone)] +pub struct LibraryAnalysisBackfillSession { + pub server_index_key: String, + pub library_server_id: String, + pub server_url: String, + pub username: String, + pub password: String, + pub workers: u32, +} + +pub struct LibraryAnalysisBackfillWorker { + pub enabled: AtomicBool, + session: Mutex>, + cursor: Mutex>, + completed_total: Mutex>, + exhausted_streak: Mutex, +} + +impl LibraryAnalysisBackfillWorker { + pub fn new() -> Self { + Self { + enabled: AtomicBool::new(false), + session: Mutex::new(None), + cursor: Mutex::new(None), + completed_total: Mutex::new(None), + exhausted_streak: Mutex::new(0), + } + } + + pub async fn set_session(&self, enabled: bool, session: Option) { + self.enabled.store(enabled, Ordering::Relaxed); + *self.session.lock().await = session; + if !enabled { + *self.cursor.lock().await = None; + *self.completed_total.lock().await = None; + *self.exhausted_streak.lock().await = 0; + } + } +} + +async fn session_still_focused( + worker: &LibraryAnalysisBackfillWorker, + expected: &LibraryAnalysisBackfillSession, +) -> bool { + if !worker.enabled.load(Ordering::Relaxed) { + return false; + } + worker + .session + .lock() + .await + .as_ref() + .is_some_and(|s| s.server_index_key == expected.server_index_key) +} + +fn sync_blocks_backfill(store: &psysonic_library::store::LibraryStore, server_id: &str) -> bool { + use psysonic_library::repos::sync_state::SyncStateRepository; + let repo = SyncStateRepository::new(store); + match repo.get_sync_phase(server_id, "") { + Ok(Some(phase)) => phase == "initial_sync" || phase == "probing", + _ => false, + } +} + +fn session_matches_server(session: &LibraryAnalysisBackfillSession, server_id: &str) -> bool { + server_id == session.server_index_key || server_id == session.library_server_id +} + +struct CoordinatorTick { + sleep_ms: u64, +} + +async fn run_coordinator_forever(app: AppHandle, worker: Arc) { + loop { + let tick = coordinator_tick(&app, worker.as_ref()).await; + tokio::time::sleep(Duration::from_millis(tick.sleep_ms)).await; + } +} + +async fn coordinator_tick( + app: &AppHandle, + worker: &LibraryAnalysisBackfillWorker, +) -> CoordinatorTick { + if !worker.enabled.load(Ordering::Relaxed) { + return CoordinatorTick { sleep_ms: STEADY_POLL_MS }; + } + + let session = worker.session.lock().await.clone(); + let Some(session) = session else { + return CoordinatorTick { sleep_ms: STEADY_POLL_MS }; + }; + + if !session_still_focused(worker, &session).await { + return CoordinatorTick { sleep_ms: STEADY_POLL_MS }; + } + + analysis_set_pipeline_parallelism(session.workers as usize); + + let runtime = match app.try_state::() { + Some(r) => r, + None => return CoordinatorTick { sleep_ms: READY_POLL_MS }, + }; + + if let Some(done_total) = *worker.completed_total.lock().await { + let store = runtime.store.clone(); + let lib_id = session.library_server_id.clone(); + let still_same = tauri::async_runtime::spawn_blocking(move || { + TrackRepository::new(&store) + .count_live_tracks(&lib_id) + .map(|n| n == done_total) + }) + .await + .ok() + .and_then(|r| r.ok()) + .unwrap_or(false); + if still_same { + return CoordinatorTick { + sleep_ms: COMPLETED_RECHECK_MS, + }; + } + *worker.completed_total.lock().await = None; + *worker.cursor.lock().await = None; + *worker.exhausted_streak.lock().await = 0; + } + + while sync_blocks_backfill(&runtime.store, &session.library_server_id) { + if !worker.enabled.load(Ordering::Relaxed) { + return CoordinatorTick { sleep_ms: SYNC_WAIT_MS }; + } + tokio::time::sleep(Duration::from_millis(SYNC_WAIT_MS)).await; + } + + let store = runtime.store.clone(); + let lib_id = session.library_server_id.clone(); + let ready = tauri::async_runtime::spawn_blocking(move || library_server_is_ready(&store, &lib_id)) + .await + .ok() + .and_then(|r| r.ok()) + .unwrap_or(false); + if !ready { + return CoordinatorTick { sleep_ms: READY_POLL_MS }; + } + + let stats = analysis_pipeline_queue_stats(); + let counts = PipelineBacklogCounts { + http_queued: stats.http_queued as u32, + http_download_active: stats.http_download_active as u32, + cpu_queued: stats.cpu_queued as u32, + cpu_decode_active: stats.cpu_decode_active as u32, + }; + + if !library_backfill_needs_top_up(counts, session.workers) { + return CoordinatorTick { sleep_ms: STEADY_POLL_MS }; + } + + let fetch_limit = library_backfill_top_up_limit(counts, session.workers); + if fetch_limit == 0 { + return CoordinatorTick { sleep_ms: TOP_UP_POLL_MS }; + } + + let cursor = worker.cursor.lock().await.clone(); + let app_for_batch = app.clone(); + let lib_id = session.library_server_id.clone(); + let batch: Option = tauri::async_runtime::spawn_blocking( + move || { + let runtime = app_for_batch + .try_state::() + .ok_or_else(|| "LibraryRuntime not available".to_string())?; + collect_analysis_backfill_batch( + &app_for_batch, + &runtime, + lib_id.trim(), + cursor.as_deref().filter(|s| !s.is_empty()), + Some(fetch_limit), + ) + }, + ) + .await + .ok() + .and_then(|r| r.ok()); + + let Some(batch) = batch else { + return CoordinatorTick { sleep_ms: TOP_UP_POLL_MS }; + }; + + *worker.cursor.lock().await = batch.next_cursor.clone(); + + let enqueued_count = batch.track_ids.len(); + let track_ids = batch.track_ids.clone(); + let index_key = session.server_index_key.clone(); + let server_url = session.server_url.clone(); + let username = session.username.clone(); + let password = session.password.clone(); + let app_for_enqueue = app.clone(); + + let _ = tauri::async_runtime::spawn_blocking(move || { + for track_id in &track_ids { + let url = build_stream_view_url(&server_url, &username, &password, track_id); + let _ = enqueue_seed_from_url( + &app_for_enqueue, + track_id, + &url, + Some(index_key.as_str()), + Some(AnalysisBackfillPriority::Low), + false, + ); + } + }) + .await; + + if enqueued_count > 0 { + *worker.exhausted_streak.lock().await = 0; + } + + if batch.exhausted { + let app_for_progress = app.clone(); + let lib_id = session.library_server_id.clone(); + let progress = tauri::async_runtime::spawn_blocking(move || { + let runtime = app_for_progress + .try_state::() + .ok_or_else(|| "LibraryRuntime not available".to_string())?; + collect_analysis_progress(&app_for_progress, &runtime, lib_id.trim()) + }) + .await + .ok() + .and_then(|r| r.ok()); + + let pending = progress.as_ref().map(|p| p.pending_tracks).unwrap_or(-1); + if pending <= 0 && enqueued_count == 0 { + let mut streak = worker.exhausted_streak.lock().await; + *streak += 1; + if *streak >= EXHAUSTED_DONE_STREAK { + if let Some(p) = progress { + *worker.completed_total.lock().await = Some(p.total_tracks); + } + *worker.cursor.lock().await = None; + return CoordinatorTick { + sleep_ms: COMPLETED_RECHECK_MS, + }; + } + } else { + *worker.exhausted_streak.lock().await = 0; + } + if pending > 0 { + *worker.cursor.lock().await = None; + return CoordinatorTick { + sleep_ms: EXHAUSTED_PENDING_RESCAN_MS, + }; + } + *worker.cursor.lock().await = None; + return CoordinatorTick { + sleep_ms: EXHAUSTED_PAUSE_MS, + }; + } + + CoordinatorTick { sleep_ms: TOP_UP_POLL_MS } +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct SyncIdlePayload { + server_id: String, + ok: bool, +} + +fn on_sync_idle(app: &AppHandle, payload: SyncIdlePayload) { + if !payload.ok { + return; + } + let app = app.clone(); + tauri::async_runtime::spawn(async move { + let worker = match app.try_state::>() { + Some(w) => w.inner().clone(), + None => return, + }; + if !worker.enabled.load(Ordering::Relaxed) { + return; + } + let session = worker.session.lock().await.clone(); + let Some(session) = session else { + return; + }; + if !session_matches_server(&session, &payload.server_id) { + return; + } + *worker.cursor.lock().await = None; + *worker.exhausted_streak.lock().await = 0; + }); +} + +pub fn setup_library_sync_idle_listener(app: &AppHandle) { + let app_handle = app.clone(); + let _ = app.listen(LibrarySyncProgressPayload::IDLE_EVENT_NAME, move |event| { + let Ok(payload) = serde_json::from_str::(event.payload()) else { + return; + }; + on_sync_idle(&app_handle, payload); + }); +} + +pub fn spawn_coordinator(app: &AppHandle, worker: Arc) { + let app = app.clone(); + tauri::async_runtime::spawn(async move { + run_coordinator_forever(app, worker).await; + }); +} diff --git a/src/api/analysis.ts b/src/api/analysis.ts index d28c2e88..90bcdca7 100644 --- a/src/api/analysis.ts +++ b/src/api/analysis.ts @@ -28,12 +28,6 @@ export interface AnalysisPipelineQueueStatsDto { cpuDecodeActiveLow: number; } -export interface LibraryAnalysisBackfillBatchDto { - trackIds: string[]; - nextCursor: string | null; - exhausted: boolean; -} - export interface LibraryAnalysisProgressDto { totalTracks: number; pendingTracks: number; @@ -52,8 +46,6 @@ export interface AnalysisDeleteServerReportDto { loudness: number; } -export const LIBRARY_ANALYSIS_BACKFILL_BATCH_SIZE = 20; - function serverIndexKeyForId(serverId: string): string { const server = useAuthStore.getState().servers.find(s => s.id === serverId); if (!server) return serverId; @@ -68,19 +60,6 @@ export function analysisGetPipelineQueueStats(): Promise('analysis_get_pipeline_queue_stats'); } -export function libraryAnalysisBackfillBatch( - serverId: string, - cursor?: string | null, - limit = LIBRARY_ANALYSIS_BACKFILL_BATCH_SIZE, -): Promise { - const indexKey = serverIndexKeyForId(serverId); - return invoke('library_analysis_backfill_batch', { - serverId: indexKey, - cursor: cursor ?? null, - limit, - }); -} - export function libraryAnalysisProgress( serverId: string, ): Promise { @@ -157,3 +136,21 @@ export function analysisEnqueueSeedFromUrl( const indexKey = serverIndexKeyForId(serverId); return invoke('analysis_enqueue_seed_from_url', { trackId, url, serverId: indexKey, priority }); } + +export type LibraryAnalysisBackfillConfigureArgs = { + enabled: boolean; + serverIndexKey: string; + libraryServerId: string; + serverUrl: string; + username: string; + password: string; + workers: number; +}; + +/** Start/stop native library analysis backfill (advanced strategy only). */ +export function libraryAnalysisBackfillConfigure( + args: LibraryAnalysisBackfillConfigureArgs, +): Promise { + // Flat payload โ€” same as `library_cover_backfill_configure` (not `{ args: โ€ฆ }`). + return invoke('library_analysis_backfill_configure', args); +} diff --git a/src/config/settingsCredits.ts b/src/config/settingsCredits.ts index a9bf6455..b3e14e59 100644 --- a/src/config/settingsCredits.ts +++ b/src/config/settingsCredits.ts @@ -135,6 +135,7 @@ const CONTRIBUTOR_ENTRIES = [ 'Albums: combined browse filters (genre/year/favorites/lossless/compilations), session restore from album detail, favorites reconcile via local index (PR #876)', 'Artist detail: sort albums by year (newest/oldest) in the Albums section (PR #877)', 'Cover art: Windows thumbnails, tier fallback, PNG decode, Subsonic coverArt id resolution (PR #878)', + 'Analytics: native advanced library backfill coordinator โ€” UI stays responsive on large libraries (PR #881)', ], }, { diff --git a/src/hooks/useLibraryAnalysisBackfill.ts b/src/hooks/useLibraryAnalysisBackfill.ts index 07af7f96..2b1b9f74 100644 --- a/src/hooks/useLibraryAnalysisBackfill.ts +++ b/src/hooks/useLibraryAnalysisBackfill.ts @@ -1,64 +1,38 @@ -import { useEffect, useRef } from 'react'; -import { buildStreamUrlForServer } from '../api/subsonicStreamUrl'; +import { useEffect } from 'react'; import { - analysisEnqueueSeedFromUrl, - analysisGetPipelineQueueStats, analysisSetPipelineParallelism, - libraryAnalysisBackfillBatch, - libraryCountLiveTracks, - libraryAnalysisProgress, + libraryAnalysisBackfillConfigure, } from '../api/analysis'; +import { librarySqlServerId } from '../api/coverCache'; import { useAuthStore } from '../store/authStore'; import { useAnalysisStrategyStore } from '../store/analysisStrategyStore'; import { DEFAULT_ADVANCED_PARALLELISM } from '../utils/library/analysisStrategy'; -import { - libraryBackfillNeedsTopUp, - libraryBackfillTopUpLimit, -} from '../utils/library/libraryAnalysisBackfillPolicy'; -import { libraryIsReady } from '../utils/library/libraryReady'; +import { serverIndexKeyForProfile } from '../utils/server/serverIndexKey'; -const TOP_UP_POLL_MS = 500; -const STEADY_POLL_MS = 2000; -const READY_POLL_MS = 5000; -const EXHAUSTED_PAUSE_MS = 15_000; -const COMPLETED_RECHECK_MS = 5 * 60_000; -/** Consecutive exhausted scans with zero enqueue before treating the library as done. */ -const EXHAUSTED_DONE_STREAK = 2; - -const EMPTY_PIPELINE_STATS = { - pipelineWorkers: 1, - httpQueued: 0, - httpQueuedHigh: 0, - httpQueuedMiddle: 0, - httpQueuedLow: 0, - httpDownloadActive: 0, - httpDownloadActiveHigh: 0, - httpDownloadActiveMiddle: 0, - httpDownloadActiveLow: 0, - cpuQueued: 0, - cpuQueuedHigh: 0, - cpuQueuedMiddle: 0, - cpuQueuedLow: 0, - cpuDecodeActive: 0, - cpuDecodeActiveHigh: 0, - cpuDecodeActiveMiddle: 0, - cpuDecodeActiveLow: 0, -}; +const DISABLED_CONFIGURE = { + enabled: false, + serverIndexKey: '', + libraryServerId: '', + serverUrl: '', + username: '', + password: '', + workers: DEFAULT_ADVANCED_PARALLELISM, +} as const; /** - * Advanced analytics strategy: keep the HTTP backfill backlog above worker count - * (watermark target โ‰ˆ workers ร— 3) so parallel downloads stay busy. Library - * tracks enqueue at low priority; playback tiers stay ahead in Rust. + * Advanced analytics strategy: native coordinator in Rust (see `library_analysis_backfill`). + * Webview only passes session + worker count โ€” no planning loop or batch IPC. */ export function useLibraryAnalysisBackfill(enabled = true): void { const activeServerId = useAuthStore(s => s.activeServerId); + const server = useAuthStore(s => + s.activeServerId ? s.servers.find(srv => srv.id === s.activeServerId) : undefined, + ); + const getBaseUrl = useAuthStore(s => s.getBaseUrl); const strategy = useAnalysisStrategyStore(s => s.getStrategyForServer(activeServerId)); const advancedParallelism = useAnalysisStrategyStore( s => s.getAdvancedParallelismForServer(activeServerId), ); - const cursorRef = useRef(null); - const completedTotalRef = useRef(null); - const exhaustedIdleStreakRef = useRef(0); useEffect(() => { if (!enabled) return; @@ -68,110 +42,38 @@ export function useLibraryAnalysisBackfill(enabled = true): void { }, [strategy, advancedParallelism, enabled]); useEffect(() => { - if (!enabled || strategy !== 'advanced' || !activeServerId) return; - - let cancelled = false; - const serverId = activeServerId; - - void (async () => { - while (!cancelled) { - if (completedTotalRef.current !== null) { - const totalTracks = await libraryCountLiveTracks(serverId).catch(() => null); - if (!Number.isFinite(totalTracks)) { - await new Promise(r => setTimeout(r, COMPLETED_RECHECK_MS)); - continue; - } - if (totalTracks === completedTotalRef.current) { - await new Promise(r => setTimeout(r, COMPLETED_RECHECK_MS)); - continue; - } - completedTotalRef.current = null; - cursorRef.current = null; - exhaustedIdleStreakRef.current = 0; - } - - if (!(await libraryIsReady(serverId))) { - await new Promise(r => setTimeout(r, READY_POLL_MS)); - continue; - } - - const workers = advancedParallelism; - const stats = await analysisGetPipelineQueueStats().catch( - () => EMPTY_PIPELINE_STATS, - ); - - if (!libraryBackfillNeedsTopUp(stats, workers)) { - await new Promise(r => setTimeout(r, STEADY_POLL_MS)); - continue; - } - - const fetchLimit = libraryBackfillTopUpLimit(stats, workers); - if (fetchLimit <= 0) { - await new Promise(r => setTimeout(r, TOP_UP_POLL_MS)); - continue; - } - - const batch = await libraryAnalysisBackfillBatch( - serverId, - cursorRef.current, - fetchLimit, - ).catch(() => null); - - if (!batch || cancelled) { - await new Promise(r => setTimeout(r, TOP_UP_POLL_MS)); - continue; - } - - cursorRef.current = batch.nextCursor; - - await Promise.all( - batch.trackIds.map(trackId => - analysisEnqueueSeedFromUrl( - trackId, - buildStreamUrlForServer(serverId, trackId), - serverId, - 'low', - ).catch(() => { - /* Rust skips completed tracks */ - }), - ), - ); - - if (cancelled) return; - - if (batch.trackIds.length > 0) { - exhaustedIdleStreakRef.current = 0; - } - - if (batch.exhausted) { - cursorRef.current = null; - const progress = await libraryAnalysisProgress(serverId).catch(() => null); - const pending = progress?.pendingTracks ?? -1; - if (pending <= 0 && batch.trackIds.length === 0) { - exhaustedIdleStreakRef.current += 1; - if (exhaustedIdleStreakRef.current >= EXHAUSTED_DONE_STREAK && progress) { - completedTotalRef.current = progress.totalTracks; - await new Promise(r => setTimeout(r, COMPLETED_RECHECK_MS)); - continue; - } - } else { - exhaustedIdleStreakRef.current = 0; - } - if (pending > 0) { - continue; - } - await new Promise(r => setTimeout(r, EXHAUSTED_PAUSE_MS)); - } else if (batch.trackIds.length === 0) { - await new Promise(r => setTimeout(r, TOP_UP_POLL_MS)); - } - } - })(); - - return () => { - cancelled = true; - cursorRef.current = null; - completedTotalRef.current = null; - exhaustedIdleStreakRef.current = 0; + const disable = () => { + void libraryAnalysisBackfillConfigure(DISABLED_CONFIGURE); }; - }, [strategy, activeServerId, advancedParallelism, enabled]); + + if (!enabled || strategy !== 'advanced' || !activeServerId || !server) { + disable(); + return disable; + } + + const indexKey = serverIndexKeyForProfile(server); + const baseUrl = getBaseUrl(); + void libraryAnalysisBackfillConfigure({ + enabled: true, + serverIndexKey: indexKey, + libraryServerId: librarySqlServerId(activeServerId), + serverUrl: baseUrl, + username: server.username, + password: server.password, + workers: advancedParallelism, + }).catch(() => { + /* coordinator optional; avoid unhandled rejection noise in release */ + }); + + return disable; + }, [ + enabled, + strategy, + activeServerId, + server?.url, + server?.username, + server?.password, + advancedParallelism, + getBaseUrl, + ]); } diff --git a/src/store/audioListenerSetup/audioEngineListeners.ts b/src/store/audioListenerSetup/audioEngineListeners.ts index e197812b..b096b3af 100644 --- a/src/store/audioListenerSetup/audioEngineListeners.ts +++ b/src/store/audioListenerSetup/audioEngineListeners.ts @@ -15,6 +15,7 @@ import { hasStableLoudness, setCachedLoudnessGain, } from '../loudnessGainCache'; +import { isTrackInsideLoudnessBackfillWindow } from '../loudnessBackfillWindow'; import { refreshLoudnessForTrack } from '../loudnessRefresh'; import { emitNormalizationDebug } from '../normalizationDebug'; import { @@ -96,8 +97,19 @@ export function setupAudioEngineListeners(): () => void { emitNormalizationDebug('backfill:applied', { trackId: currentId }); return; } - // Backfill finished for another id (e.g. next in queue): refresh loudness cache only - // so the cached gain is ready before `audio_play` / gapless chain. + // Library aggressive backfill completes thousands of tracks โ€” only warm loudness + // for the playback window (current + next N). Skip IPC for the rest. + const live = usePlayerStore.getState(); + if ( + !isTrackInsideLoudnessBackfillWindow( + payloadTrackId, + live.queueItems, + live.queueIndex, + live.currentTrack, + ) + ) { + return; + } void refreshLoudnessForTrack(payloadTrackId, { syncPlayingEngine: false }); emitNormalizationDebug('backfill:applied', { trackId: payloadTrackId }); }), diff --git a/src/utils/library/libraryAnalysisBackfillPolicy.test.ts b/src/utils/library/libraryAnalysisBackfillPolicy.test.ts deleted file mode 100644 index 8d60b7cd..00000000 --- a/src/utils/library/libraryAnalysisBackfillPolicy.test.ts +++ /dev/null @@ -1,48 +0,0 @@ -import { describe, it, expect } from 'vitest'; -import { - computeLibraryBackfillTargetDepth, - libraryBackfillNeedsTopUp, - libraryBackfillPipelineBacklog, - libraryBackfillTopUpLimit, -} from './libraryAnalysisBackfillPolicy'; - -describe('libraryAnalysisBackfillPolicy', () => { - it('targets workers ร— 3 with floor and cap', () => { - expect(computeLibraryBackfillTargetDepth(1)).toBe(8); - expect(computeLibraryBackfillTargetDepth(4)).toBe(12); - expect(computeLibraryBackfillTargetDepth(8)).toBe(24); - expect(computeLibraryBackfillTargetDepth(100)).toBe(240); - }); - - const zeroCpu = { cpuQueued: 0, cpuDecodeActive: 0 }; - - it('measures backlog as HTTP plus CPU seed pipeline load', () => { - expect( - libraryBackfillPipelineBacklog({ - httpQueued: 5, - httpDownloadActive: 3, - ...zeroCpu, - }), - ).toBe(8); - expect( - libraryBackfillPipelineBacklog({ - httpQueued: 0, - httpDownloadActive: 0, - cpuQueued: 4, - cpuDecodeActive: 2, - }), - ).toBe(6); - }); - - it('requests top-up while backlog stays below target', () => { - const stats = { httpQueued: 2, httpDownloadActive: 1, ...zeroCpu }; - expect(libraryBackfillNeedsTopUp(stats, 8)).toBe(true); - expect(libraryBackfillTopUpLimit(stats, 8)).toBe(20); - }); - - it('stops top-up when backlog meets target', () => { - const stats = { httpQueued: 20, httpDownloadActive: 4, ...zeroCpu }; - expect(libraryBackfillNeedsTopUp(stats, 8)).toBe(false); - expect(libraryBackfillTopUpLimit(stats, 8)).toBe(0); - }); -}); diff --git a/src/utils/library/libraryAnalysisBackfillPolicy.ts b/src/utils/library/libraryAnalysisBackfillPolicy.ts deleted file mode 100644 index 966c7f14..00000000 --- a/src/utils/library/libraryAnalysisBackfillPolicy.ts +++ /dev/null @@ -1,41 +0,0 @@ -import type { AnalysisPipelineQueueStatsDto } from '../../api/analysis'; -import { LIBRARY_ANALYSIS_BACKFILL_BATCH_SIZE } from '../../api/analysis'; - -/** Target backlog โ‰ˆ workers ร— multiplier (min floor, max cap). */ -export const LIBRARY_BACKLOG_DEPTH_MULTIPLIER = 3; -export const LIBRARY_BACKLOG_MIN = 8; -export const LIBRARY_BACKLOG_MAX = 240; - -export function computeLibraryBackfillTargetDepth(workers: number): number { - const w = Math.max(1, Math.round(workers)); - return Math.min( - LIBRARY_BACKLOG_MAX, - Math.max(LIBRARY_BACKLOG_MIN, w * LIBRARY_BACKLOG_DEPTH_MULTIPLIER), - ); -} - -type PipelineBacklogStats = Pick< - AnalysisPipelineQueueStatsDto, - 'httpQueued' | 'httpDownloadActive' | 'cpuQueued' | 'cpuDecodeActive' ->; - -/** HTTP + in-flight CPU seed jobs (decode backpressure must not look like an empty pipeline). */ -export function libraryBackfillPipelineBacklog(stats: PipelineBacklogStats): number { - return ( - stats.httpQueued - + stats.httpDownloadActive - + stats.cpuQueued - + stats.cpuDecodeActive - ); -} - -export function libraryBackfillNeedsTopUp(stats: PipelineBacklogStats, workers: number): boolean { - return libraryBackfillPipelineBacklog(stats) < computeLibraryBackfillTargetDepth(workers); -} - -export function libraryBackfillTopUpLimit(stats: PipelineBacklogStats, workers: number): number { - const target = computeLibraryBackfillTargetDepth(workers); - const deficit = target - libraryBackfillPipelineBacklog(stats); - if (deficit <= 0) return 0; - return Math.min(LIBRARY_ANALYSIS_BACKFILL_BATCH_SIZE, deficit); -}