diff --git a/CHANGELOG.md b/CHANGELOG.md index 42d7309f..a0a3076e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -132,6 +132,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * **False-positive mitigation:** the watchdog now arms only after a long poll gap (sleep/resume-like condition) and logs arm/clear/trigger decisions, reducing unexpected stream reopens during normal playback. * **Card hover stability:** removed vertical lift on album/artist/base cards to avoid pointer-edge pulsation, kept artwork zoom smooth, and dropped per-card GPU layer hints that could regress software-composited Linux paths. +### Analysis queue control — prune stale backfill jobs and cap warmup window + +**By [@cucadmuh](https://github.com/cucadmuh), PR [#480](https://github.com/Psychotoxical/psysonic/pull/480)** + +* Added a queue-prune path for pending analysis work so stale `http_backfill` / `cpu_seed` jobs are dropped when tracks leave the active playback queue. +* Limited loudness backfill warmup to the current track plus the next 5 tracks, reducing runaway analysis scheduling from large bulk queue updates. +* Added debug counters for prune results to make queue-pressure behavior visible during diagnostics. + ## [1.45.0] - 2026-05-04 ## Added diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index a68bf001..376d647a 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -10,7 +10,7 @@ mod lib_commands; #[cfg(target_os = "windows")] mod taskbar_win; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, Mutex, OnceLock, RwLock}; use std::sync::atomic::{AtomicBool, Ordering}; @@ -127,6 +127,13 @@ impl AnalysisBackfillQueueState { AnalysisBackfillEnqueueKind::NewBack } } + + fn prune_queued_not_in(&mut self, keep_track_ids: &HashSet<&str>) -> usize { + let before = self.deque.len(); + self.deque + .retain(|(track_id, _)| keep_track_ids.contains(track_id.as_str())); + before.saturating_sub(self.deque.len()) + } } struct AnalysisBackfillShared { @@ -300,6 +307,27 @@ impl AnalysisCpuSeedQueueState { }; (kind, done_rx) } + + fn prune_queued_not_in(&mut self, keep_track_ids: &HashSet<&str>) -> (usize, usize) { + let mut kept = VecDeque::with_capacity(self.deque.len()); + let mut removed_jobs = 0usize; + let mut removed_waiters = 0usize; + while let Some(job) = self.deque.pop_front() { + if keep_track_ids.contains(job.track_id.as_str()) { + kept.push_back(job); + continue; + } + removed_jobs += 1; + removed_waiters += job.waiters.len(); + for tx in job.waiters { + let _ = tx.send(Err( + "cpu-seed pruned: track no longer in playback queue".to_string(), + )); + } + } + self.deque = kept; + (removed_jobs, removed_waiters) + } } struct AnalysisCpuSeedShared { @@ -920,6 +948,7 @@ pub fn run() { analysis_delete_loudness_for_track, analysis_delete_all_waveforms, analysis_enqueue_seed_from_url, + analysis_prune_pending_to_track_ids, download_track_offline, delete_offline_track, get_offline_cache_size, diff --git a/src-tauri/src/lib_commands/app_api/analysis.rs b/src-tauri/src/lib_commands/app_api/analysis.rs index 95dbd417..54d9f5a3 100644 --- a/src-tauri/src/lib_commands/app_api/analysis.rs +++ b/src-tauri/src/lib_commands/app_api/analysis.rs @@ -1,4 +1,5 @@ use super::*; +use std::collections::HashSet; #[tauri::command] pub(crate) fn analysis_get_waveform( @@ -174,3 +175,70 @@ pub(crate) fn analysis_enqueue_seed_from_url( } Ok(()) } + +#[derive(Debug, Clone, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct AnalysisPrunePendingResult { + pub keep_count: usize, + pub http_removed: usize, + pub cpu_removed_jobs: usize, + pub cpu_removed_waiters: usize, +} + +/// Prunes pending analysis work for tracks no longer present in the playback queue. +/// +/// Keeps currently-running jobs untouched; only queued (not-yet-started) jobs are removed. +#[tauri::command] +pub(crate) fn analysis_prune_pending_to_track_ids( + track_ids: Vec, +) -> Result { + let mut normalized: Vec = Vec::with_capacity(track_ids.len()); + let mut seen = HashSet::new(); + for raw in track_ids { + let tid = raw.trim(); + if tid.is_empty() { + continue; + } + if seen.insert(tid.to_string()) { + normalized.push(tid.to_string()); + } + } + let keep_track_ids: HashSet<&str> = normalized.iter().map(|s| s.as_str()).collect(); + + let http_removed = if let Some(shared) = ANALYSIS_BACKFILL.get() { + let mut st = shared + .state + .lock() + .map_err(|_| "analysis backfill lock poisoned".to_string())?; + st.prune_queued_not_in(&keep_track_ids) + } else { + 0 + }; + + let (cpu_removed_jobs, cpu_removed_waiters) = if let Some(shared) = ANALYSIS_CPU_SEED.get() { + let mut st = shared + .state + .lock() + .map_err(|_| "analysis cpu-seed lock poisoned".to_string())?; + st.prune_queued_not_in(&keep_track_ids) + } else { + (0, 0) + }; + + if http_removed > 0 || cpu_removed_jobs > 0 { + crate::app_deprintln!( + "[analysis] pruned pending queues keep={} removed_http={} removed_cpu_jobs={} removed_cpu_waiters={}", + keep_track_ids.len(), + http_removed, + cpu_removed_jobs, + cpu_removed_waiters + ); + } + + Ok(AnalysisPrunePendingResult { + keep_count: keep_track_ids.len(), + http_removed, + cpu_removed_jobs, + cpu_removed_waiters, + }) +} diff --git a/src/hotCachePrefetch.ts b/src/hotCachePrefetch.ts index 927bba64..a4e7ad28 100644 --- a/src/hotCachePrefetch.ts +++ b/src/hotCachePrefetch.ts @@ -73,6 +73,56 @@ let debounceTimer: ReturnType | null = null; let graceEvictTimer: ReturnType | null = null; const pendingQueue: PrefetchJob[] = []; let workerRunning = false; +let analysisPruneTimer: ReturnType | null = null; +let lastAnalysisPruneSig = ''; +const ANALYSIS_PRUNE_DEBOUNCE_MS = 1200; + +type AnalysisPrunePendingResult = { + keepCount: number; + httpRemoved: number; + cpuRemovedJobs: number; + cpuRemovedWaiters: number; +}; + +function scheduleAnalysisQueuePruneFromPlaybackQueue(): void { + const { queue, currentTrack } = usePlayerStore.getState(); + const keepTrackIds: string[] = []; + const seen = new Set(); + const pushId = (id: string | undefined | null) => { + if (!id) return; + const tid = id.trim(); + if (!tid || seen.has(tid)) return; + seen.add(tid); + keepTrackIds.push(tid); + }; + pushId(currentTrack?.id); + for (const track of queue) { + pushId(track.id); + if (keepTrackIds.length >= 1000) break; + } + const sig = JSON.stringify(keepTrackIds); + if (sig === lastAnalysisPruneSig) return; + lastAnalysisPruneSig = sig; + if (analysisPruneTimer) { + clearTimeout(analysisPruneTimer); + analysisPruneTimer = null; + } + analysisPruneTimer = setTimeout(() => { + analysisPruneTimer = null; + void invoke('analysis_prune_pending_to_track_ids', { trackIds: keepTrackIds }) + .then(result => { + if (!result) return; + hotCacheFrontendDebug({ + event: 'analysis-prune', + keepCount: result.keepCount, + removedHttp: result.httpRemoved, + removedCpuJobs: result.cpuRemovedJobs, + removedCpuWaiters: result.cpuRemovedWaiters, + }); + }) + .catch(() => {}); + }, ANALYSIS_PRUNE_DEBOUNCE_MS); +} function debounceMs(): number { const s = useAuthStore.getState().hotCacheDebounceSec; @@ -332,6 +382,7 @@ export function initHotCachePrefetch(): () => void { const onlyIndexMoved = q === lastQueueRef && i !== lastQueueIndex; lastQueueRef = q; lastQueueIndex = i; + scheduleAnalysisQueuePruneFromPlaybackQueue(); if (onlyIndexMoved && i > prevIdx && prevIdx >= 0 && Array.isArray(prevQ)) { const left = (prevQ as Track[])[prevIdx]; const a = useAuthStore.getState(); @@ -393,6 +444,7 @@ export function initHotCachePrefetch(): () => void { }); void replanNow(); + scheduleAnalysisQueuePruneFromPlaybackQueue(); return () => { unsubPlayer(); @@ -401,6 +453,9 @@ export function initHotCachePrefetch(): () => void { debounceTimer = null; if (graceEvictTimer) clearTimeout(graceEvictTimer); graceEvictTimer = null; + if (analysisPruneTimer) clearTimeout(analysisPruneTimer); + analysisPruneTimer = null; + lastAnalysisPruneSig = ''; pendingQueue.length = 0; clearHotCachePreviousGrace(); }; diff --git a/src/pages/Settings.tsx b/src/pages/Settings.tsx index 52a273a7..21e3b777 100644 --- a/src/pages/Settings.tsx +++ b/src/pages/Settings.tsx @@ -209,6 +209,7 @@ const CONTRIBUTORS = [ 'Shortcuts: action registry, dynamic CLI help, 9 new input targets + F1 help binding (PR #435)', 'Environment upgrade & hot-cache playback — replay via RAM/disk on same-track and queue-end resume, playback-source icon stays correct after resume/undo/gapless, sidebar new-releases 500-id cap merge, Windows tray double-click fix, lazy-loaded routes, rodio 0.22 migration (PR #463)', 'Audio: post-sleep stream recovery (Windows + Linux) with poll-gap-armed stall watchdog; preview seekbar freeze + anti-jump on preview end; remove card hover lift and per-card GPU compositing hints (PR #476)', + 'Analysis queue control: prune stale http-backfill / cpu-seed jobs when tracks leave the playback queue, cap loudness backfill warmup to current + next 5 tracks, plus debug counters for diagnostics (PR #480)', ], }, { diff --git a/src/store/playerStore.ts b/src/store/playerStore.ts index f011646f..cdab27c0 100644 --- a/src/store/playerStore.ts +++ b/src/store/playerStore.ts @@ -1110,6 +1110,14 @@ async function runRefreshLoudnessForTrack(trackId: string, syncEngine: boolean): if (auth.normalizationEngine === 'loudness' && !analysisBackfillInFlightByTrackId[trackId] && attempts < MAX_BACKFILL_ATTEMPTS_PER_TRACK) { + if (!isTrackInsideLoudnessBackfillWindow(trackId)) { + emitNormalizationDebug('backfill:skipped-outside-window', { + trackId, + queueIndex: usePlayerStore.getState().queueIndex, + aheadWindow: LOUDNESS_BACKFILL_WINDOW_AHEAD, + }); + return; + } analysisBackfillInFlightByTrackId[trackId] = true; analysisBackfillAttemptsByTrackId[trackId] = attempts + 1; const url = buildStreamUrl(trackId); @@ -1159,25 +1167,41 @@ async function runRefreshLoudnessForTrack(trackId: string, syncEngine: boolean): } /** After bulk enqueue, warm loudness cache so gapless `audio_chain_preload` sees real gain, not only startup trim. */ -const LOUDNESS_PREFETCH_MAX_ENQUEUED_IDS = 40; +const LOUDNESS_BACKFILL_WINDOW_AHEAD = 5; + +function isTrackInsideLoudnessBackfillWindow(trackId: string): boolean { + if (!trackId) return false; + const state = usePlayerStore.getState(); + const currentId = state.currentTrack?.id; + if (currentId === trackId) return true; + if (state.queue.length === 0) return false; + const start = Math.max(0, state.queueIndex + 1); + const end = Math.min(state.queue.length, start + LOUDNESS_BACKFILL_WINDOW_AHEAD); + for (let i = start; i < end; i++) { + if (state.queue[i]?.id === trackId) return true; + } + return false; +} + +function collectLoudnessBackfillWindowTrackIds(queue: Track[], queueIndex: number, currentTrack: Track | null): string[] { + const ids = new Set(); + if (currentTrack?.id) ids.add(currentTrack.id); + const start = Math.max(0, queueIndex + 1); + const end = Math.min(queue.length, start + LOUDNESS_BACKFILL_WINDOW_AHEAD); + for (let i = start; i < end; i++) { + const tid = queue[i]?.id; + if (tid) ids.add(tid); + } + return Array.from(ids); +} function prefetchLoudnessForEnqueuedTracks( - incoming: Track[], mergedQueue: Track[], queueIndex: number, ) { if (useAuthStore.getState().normalizationEngine !== 'loudness') return; - const ids = new Set(); - const next = mergedQueue[queueIndex + 1]; - if (next?.id) ids.add(next.id); - let n = 0; - for (const t of incoming) { - if (n >= LOUDNESS_PREFETCH_MAX_ENQUEUED_IDS) break; - if (t?.id) { - ids.add(t.id); - n++; - } - } + const currentTrack = usePlayerStore.getState().currentTrack; + const ids = collectLoudnessBackfillWindowTrackIds(mergedQueue, queueIndex, currentTrack); for (const id of ids) { void refreshLoudnessForTrack(id, { syncPlayingEngine: false }); } @@ -3203,7 +3227,7 @@ export const usePlayerStore = create()( ...state.queue.slice(firstAutoIdx), ]; syncQueueToServer(newQueue, state.currentTrack, state.currentTime); - prefetchLoudnessForEnqueuedTracks(tracks, newQueue, state.queueIndex); + prefetchLoudnessForEnqueuedTracks(newQueue, state.queueIndex); return { queue: newQueue }; }); }, @@ -3252,7 +3276,7 @@ export const usePlayerStore = create()( ? state.queueIndex + tracks.length : state.queueIndex; syncQueueToServer(newQueue, state.currentTrack, state.currentTime); - prefetchLoudnessForEnqueuedTracks(tracks, newQueue, newQueueIndex); + prefetchLoudnessForEnqueuedTracks(newQueue, newQueueIndex); return { queue: newQueue, queueIndex: newQueueIndex }; }); },